tessl/pypi-pypika

A SQL query builder API for Python that provides programmatic interface for constructing SQL queries.

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Core Query Operations

Name: tessl/pypi-pypika
Author: tessl

Primary interface for constructing SQL queries including SELECT, INSERT, UPDATE, DELETE, and DDL operations. The Query class provides static factory methods for all query types with fluent method chaining and immutable builder pattern.

Capabilities

SELECT Queries

Build SELECT statements with comprehensive support for joins, subqueries, aggregation, ordering, and grouping.

class Query:
    @staticmethod
    def from_(table) -> QueryBuilder:
        """Initialize SELECT query from a table."""
        
    @staticmethod
    def select(*terms) -> QueryBuilder:
        """Initialize SELECT query without specifying table (for subqueries)."""

class QueryBuilder:
    def select(self, *terms) -> QueryBuilder:
        """Specify columns to select."""
        
    def where(self, criterion) -> QueryBuilder:
        """Add WHERE conditions."""
        
    def join(self, table, how=JoinType.inner) -> JoiningQueryBuilder:
        """Add JOIN clause."""
        
    def left_join(self, table) -> JoiningQueryBuilder:
        """Add LEFT JOIN clause."""
        
    def right_join(self, table) -> JoiningQueryBuilder:
        """Add RIGHT JOIN clause."""
        
    def inner_join(self, table) -> JoiningQueryBuilder:
        """Add INNER JOIN clause."""
        
    def outer_join(self, table) -> JoiningQueryBuilder:
        """Add OUTER JOIN clause."""
        
    def cross_join(self, table) -> QueryBuilder:
        """Add CROSS JOIN clause."""
        
    def groupby(self, *terms) -> QueryBuilder:
        """Add GROUP BY clause."""
        
    def having(self, criterion) -> QueryBuilder:
        """Add HAVING clause."""
        
    def orderby(self, *terms, order=Order.asc) -> QueryBuilder:
        """Add ORDER BY clause."""
        
    def limit(self, limit) -> QueryBuilder:
        """Add LIMIT clause."""
        
    def offset(self, offset) -> QueryBuilder:
        """Add OFFSET clause."""
        
    def distinct(self) -> QueryBuilder:
        """Add DISTINCT clause."""
        
    def get_sql(self, **kwargs) -> str:
        """Generate SQL string."""

class JoiningQueryBuilder(QueryBuilder):
    def on(self, criterion) -> QueryBuilder:
        """Specify JOIN condition."""
        
    def using(self, *terms) -> QueryBuilder:
        """Specify JOIN USING clause."""

Usage Examples:

from pypika import Query, Table, Field
from pypika import JoinType, Order

users = Table('users')
orders = Table('orders')

# Basic SELECT
query = Query.from_(users).select('*')

# SELECT with specific columns
query = Query.from_(users).select(users.name, users.email, users.age)

# WHERE conditions
query = Query.from_(users).select('*').where(users.age >= 18)

# Multiple WHERE conditions
query = (Query.from_(users)
         .select('*')
         .where(users.age >= 18)
         .where(users.status == 'active'))

# JOIN operations
query = (Query.from_(users)
         .join(orders).on(users.id == orders.user_id)
         .select(users.name, orders.total))

# LEFT JOIN with complex conditions
query = (Query.from_(users)
         .left_join(orders).on((users.id == orders.user_id) & (orders.status == 'completed'))
         .select(users.name, orders.total))

# GROUP BY and HAVING
from pypika.functions import Count, Sum
query = (Query.from_(orders)
         .select(orders.user_id, Count('*').as_('order_count'), Sum(orders.total).as_('total_spent'))
         .groupby(orders.user_id)
         .having(Count('*') > 5))

# ORDER BY with multiple columns
query = (Query.from_(users)
         .select('*')
         .orderby(users.created_at, order=Order.desc)
         .orderby(users.name))

# LIMIT and OFFSET for pagination
query = (Query.from_(users)
         .select('*')
         .orderby(users.id)
         .limit(20)
         .offset(40))

# Subqueries
subquery = Query.from_(orders).select(orders.user_id).where(orders.total > 100)
query = Query.from_(users).select('*').where(users.id.isin(subquery))

INSERT Queries

Build INSERT statements supporting single and multiple value insertion, column specification, and subquery insertion.

class Query:
    @staticmethod
    def into(table) -> QueryBuilder:
        """Initialize INSERT query into table."""

class QueryBuilder:
    def insert(self, *terms) -> QueryBuilder:
        """Insert values into table."""
        
    def columns(self, *terms) -> QueryBuilder:
        """Specify target columns for INSERT."""
        
    def select(self, *terms) -> QueryBuilder:
        """Insert from SELECT subquery."""

Usage Examples:

from pypika import Query, Table

users = Table('users')

# Simple INSERT with values
query = Query.into(users).insert(1, 'John Doe', 'john@example.com')

# INSERT with column specification
query = (Query.into(users)
         .columns(users.name, users.email)
         .insert('John Doe', 'john@example.com'))

# Multiple value INSERT
query = (Query.into(users)
         .columns(users.name, users.email)
         .insert('John Doe', 'john@example.com')
         .insert('Jane Smith', 'jane@example.com'))

# INSERT from SELECT
active_users = Table('active_users')
query = (Query.into(users)
         .columns(users.name, users.email)
         .select(active_users.name, active_users.email)
         .from_(active_users)
         .where(active_users.last_login > '2023-01-01'))

UPDATE Queries

Build UPDATE statements with SET clauses, WHERE conditions, and JOIN support for complex updates.

class Query:
    @staticmethod
    def update(table) -> QueryBuilder:
        """Initialize UPDATE query on table."""

class QueryBuilder:
    def set(self, field, value) -> QueryBuilder:
        """Set field to value."""
        
    def where(self, criterion) -> QueryBuilder:
        """Add WHERE conditions."""
        
    def join(self, table, how=JoinType.inner) -> JoiningQueryBuilder:
        """Add JOIN for complex updates."""

Usage Examples:

from pypika import Query, Table
from pypika.functions import Now

users = Table('users')
orders = Table('orders')

# Simple UPDATE
query = (Query.update(users)
         .set(users.email, 'newemail@example.com')
         .where(users.id == 1))

# Multiple field UPDATE
query = (Query.update(users)
         .set(users.email, 'newemail@example.com')
         .set(users.updated_at, Now())
         .where(users.id == 1))

# UPDATE with JOIN
query = (Query.update(users)
         .join(orders).on(users.id == orders.user_id)
         .set(users.total_orders, orders.count)
         .where(orders.status == 'completed'))

# Conditional UPDATE
from pypika import Case
query = (Query.update(users)
         .set(users.status, 
              Case()
              .when(users.last_login > '2023-01-01', 'active')
              .else_('inactive'))
         .where(users.status.isnull()))

DELETE Queries

Build DELETE statements with WHERE conditions and JOIN support for complex deletion operations.

class Query:
    @staticmethod
    def from_(table) -> QueryBuilder:
        """Initialize query from table (use .delete() for DELETE operations)."""

class QueryBuilder:
    def delete(self) -> QueryBuilder:
        """Convert to DELETE query."""
        
    def where(self, criterion) -> QueryBuilder:
        """Add WHERE conditions."""
        
    def join(self, table, how=JoinType.inner) -> JoiningQueryBuilder:
        """Add JOIN for complex deletes."""

Usage Examples:

from pypika import Query, Table

users = Table('users')
orders = Table('orders')

# Simple DELETE
query = Query.from_(users).delete().where(users.id == 1)

# DELETE with complex conditions
query = (Query.from_(users)
         .delete()
         .where((users.status == 'inactive') & (users.last_login < '2022-01-01')))

# DELETE with JOIN
query = (Query.from_(users)
         .join(orders).on(users.id == orders.user_id)
         .delete()
         .where(orders.status == 'cancelled'))

DDL Operations

Data Definition Language operations for creating and dropping database objects.

class Query:
    @staticmethod
    def create_table(table) -> CreateQueryBuilder:
        """Initialize CREATE TABLE query."""
        
    @staticmethod
    def drop_table(table) -> DropQueryBuilder:
        """Initialize DROP TABLE query."""
        
    @staticmethod
    def drop_database(database) -> DropQueryBuilder:
        """Initialize DROP DATABASE query."""
        
    @staticmethod
    def drop_user(user) -> DropQueryBuilder:
        """Initialize DROP USER query."""
        
    @staticmethod
    def drop_view(view) -> DropQueryBuilder:
        """Initialize DROP VIEW query."""
        
    @staticmethod
    def Table(table_name, **kwargs) -> Table:
        """Factory method for creating Table instances."""
        
    @staticmethod
    def Tables(*names, **kwargs) -> List[Table]:
        """Factory method for creating multiple Table instances."""

class CreateQueryBuilder:
    def columns(self, *columns) -> CreateQueryBuilder:
        """Specify table columns."""
        
    def primary_key(self, *columns) -> CreateQueryBuilder:
        """Define primary key."""
        
    def foreign_key(self, column, references_table, references_column) -> CreateQueryBuilder:
        """Define foreign key constraint."""

class DropQueryBuilder:
    def if_exists(self) -> DropQueryBuilder:
        """Add IF EXISTS clause."""
        
    def cascade(self) -> DropQueryBuilder:
        """Add CASCADE clause."""

Usage Examples:

from pypika import Query, Table, Column

users = Table('users')

# CREATE TABLE
query = (Query.create_table(users)
         .columns(
             Column('id', 'INTEGER', nullable=False),
             Column('name', 'VARCHAR(100)', nullable=False),
             Column('email', 'VARCHAR(255)', nullable=False),
             Column('created_at', 'TIMESTAMP', default='CURRENT_TIMESTAMP')
         )
         .primary_key('id'))

# DROP TABLE
query = Query.drop_table(users).if_exists()

# DROP DATABASE
database = Database('test_db')
query = Query.drop_database(database).if_exists()

# DROP USER
query = Query.drop_user('test_user').if_exists()

# DROP VIEW
query = Query.drop_view('user_summary').if_exists()

# Factory methods for Table creation
users = Query.Table('users')
orders, products = Query.Tables('orders', 'products')

Common Table Expressions (CTEs)

Build queries with Common Table Expressions for complex hierarchical and recursive queries.

class Query:
    @staticmethod
    def with_(table, name) -> QueryBuilder:
        """Add Common Table Expression."""

class QueryBuilder:
    def with_(self, table, name) -> QueryBuilder:
        """Add additional CTE."""

Usage Examples:

from pypika import Query, Table, AliasedQuery

users = Table('users')
orders = Table('orders')

# Simple CTE
user_stats = AliasedQuery('user_stats')
user_stats_query = (Query.from_(orders)
                    .select(orders.user_id, Count('*').as_('order_count'))
                    .groupby(orders.user_id))

query = (Query.with_(user_stats_query, 'user_stats')
         .from_(users)
         .join(user_stats).on(users.id == user_stats.user_id)
         .select(users.name, user_stats.order_count))

# Multiple CTEs
recent_orders = AliasedQuery('recent_orders')
recent_orders_query = (Query.from_(orders)
                       .select('*')
                       .where(orders.created_at > '2023-01-01'))

query = (Query.with_(user_stats_query, 'user_stats')
         .with_(recent_orders_query, 'recent_orders')
         .from_(users)
         .join(user_stats).on(users.id == user_stats.user_id)
         .join(recent_orders).on(users.id == recent_orders.user_id)
         .select(users.name, user_stats.order_count, recent_orders.total))

Set Operations

Combine multiple queries using UNION, INTERSECT, and EXCEPT operations.

class QueryBuilder:
    def union(self, other) -> QueryBuilder:
        """UNION with another query."""
        
    def union_all(self, other) -> QueryBuilder:
        """UNION ALL with another query."""
        
    def intersect(self, other) -> QueryBuilder:
        """INTERSECT with another query."""
        
    def except_(self, other) -> QueryBuilder:
        """EXCEPT with another query."""

Usage Examples:

from pypika import Query, Table

active_users = Table('active_users')
inactive_users = Table('inactive_users')

# UNION queries
query1 = Query.from_(active_users).select(active_users.email)
query2 = Query.from_(inactive_users).select(inactive_users.email)

combined_query = query1.union(query2)

Install with Tessl CLI