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

Tables and Schema Management

Name: tessl/pypi-pypika
Author: tessl

Database schema representation with multi-level namespacing, table creation with column definitions, and field access patterns. PyPika provides comprehensive support for database schemas, tables, and columns with intuitive attribute-based field access.

Capabilities

Table Creation and Management

Core table representation supporting schema namespacing, aliasing, and field access with dynamic attribute resolution.

class Table:
    def __init__(self, name: str, schema: Optional[Union[Schema, str]] = None, 
                 alias: Optional[str] = None, query_cls: Optional[Type[Query]] = None):
        """
        Create a table reference.
        
        Parameters:
        - name: Table name
        - schema: Schema name or Schema instance
        - alias: Table alias
        - query_cls: Query class to use for operations
        """
        
    def select(self, *terms) -> QueryBuilder:
        """Start SELECT query from this table."""
        
    def update(self) -> QueryBuilder:
        """Start UPDATE query on this table."""
        
    def insert(self, *terms) -> QueryBuilder:
        """Start INSERT query into this table."""
        
    def field(self, name: str) -> Field:
        """Get field reference by name."""
        
    def as_(self, alias: str) -> Table:
        """Set table alias."""
        
    def for_(self, temporal_criterion: Criterion) -> Table:
        """Add temporal FOR clause (for temporal tables)."""
        
    def for_portion(self, period_criterion: PeriodCriterion) -> Table:
        """Add period FOR PORTION clause (for temporal tables)."""
        
    def get_table_name(self) -> str:
        """Get effective table name (alias or name)."""
        
    def get_sql(self, **kwargs) -> str:
        """Generate SQL representation."""
        
    @property
    def star(self) -> Star:
        """Get * field reference for this table."""
        
    # Dynamic field access
    def __getattr__(self, name: str) -> Field:
        """Access fields as attributes: table.field_name"""
        
    def __getitem__(self, name: str) -> Field:
        """Access fields with brackets: table['field_name']"""

Usage Examples:

from pypika import Table, Schema, Database, Query

# Simple table
users = Table('users')
print(users.get_sql())  # "users"

# Table with alias
users_alias = Table('users', alias='u')
print(users_alias.get_sql())  # "users" "u"

# Table with schema
users_schema = Table('users', schema='public')
print(users_schema.get_sql())  # "public"."users"

# Field access patterns
users = Table('users')

# Attribute access
name_field = users.name
email_field = users.email

# Bracket access  
name_field = users['name']
email_field = users['email']

# Method access
name_field = users.field('name')

# Star field
all_fields = users.star  # Represents users.*

# Query operations from table
query = users.select(users.name, users.email)
query = users.update().set(users.email, 'new@email.com')
query = users.insert('John', 'john@email.com')

Schema Management

Multi-level schema namespacing supporting nested schemas and database hierarchies.

class Schema:
    def __init__(self, name: str, parent: Optional[Schema] = None):
        """
        Create schema reference.
        
        Parameters:
        - name: Schema name
        - parent: Parent schema for nested schemas
        """
        
    def get_sql(self, quote_char: Optional[str] = None, **kwargs) -> str:
        """Generate SQL representation."""
        
    def __getattr__(self, item: str) -> Table:
        """Access tables as attributes: schema.table_name"""

class Database(Schema):
    def __getattr__(self, item: str) -> Schema:
        """Access schemas as attributes: database.schema_name"""

Usage Examples:

from pypika import Schema, Database, Table

# Simple schema
public = Schema('public')
users = Table('users', schema=public)
# Or equivalently:
users = public.users

# Nested schemas
company = Schema('company')
hr = Schema('hr', parent=company)
users = Table('users', schema=hr)
print(users.get_sql())  # "company"."hr"."users"

# Database with schemas
db = Database('mydb')
public_schema = db.public
users = public_schema.users
# Or chained:
users = db.public.users

# Multiple schema levels
db = Database('corp')
division = db.sales
region = Schema('west', parent=division)
users = Table('users', schema=region)
print(users.get_sql())  # "corp"."sales"."west"."users"

Column Definitions

Column specifications for CREATE TABLE operations with type, nullability, and default value support.

class Column:
    def __init__(self, column_name: str, column_type: Optional[str] = None,
                 nullable: Optional[bool] = None, default: Optional[Any] = None):
        """
        Define table column.
        
        Parameters:
        - column_name: Column name
        - column_type: SQL data type
        - nullable: Whether column accepts NULL (True/False/None for unspecified)
        - default: Default value or expression
        """
        
    def get_name_sql(self, **kwargs) -> str:
        """Get column name SQL."""
        
    def get_sql(self, **kwargs) -> str:
        """Get full column definition SQL."""

Usage Examples:

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

# Basic column definitions
columns = [
    Column('id', 'INTEGER', nullable=False),
    Column('name', 'VARCHAR(100)', nullable=False),
    Column('email', 'VARCHAR(255)', nullable=False),
    Column('age', 'INTEGER', nullable=True),
    Column('created_at', 'TIMESTAMP', default='CURRENT_TIMESTAMP'),
    Column('updated_at', 'TIMESTAMP', default=Now()),
    Column('status', 'VARCHAR(20)', default='active')
]

# CREATE TABLE with columns
users = Table('users')
query = (Query.create_table(users)
         .columns(*columns)
         .primary_key('id'))

print(query.get_sql())
# CREATE TABLE "users" (
#   "id" INTEGER NOT NULL,
#   "name" VARCHAR(100) NOT NULL,
#   "email" VARCHAR(255) NOT NULL,
#   "age" INTEGER NULL,
#   "created_at" TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
#   "updated_at" TIMESTAMP DEFAULT NOW(),
#   "status" VARCHAR(20) DEFAULT 'active'
# )

Table and Column Factory Functions

Utility functions for creating multiple tables and columns efficiently.

def Tables(*names: Union[Tuple[str, str], str], **kwargs) -> List[Table]:
    """
    Create multiple tables.
    
    Parameters:
    - names: Table names or (name, alias) tuples
    - kwargs: Common table options (schema, query_cls)
    
    Returns:
    List of Table instances
    """

def Columns(*names: str) -> List[Column]:
    """
    Create multiple basic columns.
    
    Parameters:
    - names: Column names
    
    Returns:
    List of Column instances
    """

Usage Examples:

from pypika import Tables, Columns, Schema

# Create multiple tables
users, orders, products = Tables('users', 'orders', 'products')

# Create tables with aliases
users, orders = Tables(('users', 'u'), ('orders', 'o'))

# Create tables with common schema
schema = Schema('public')
users, orders, products = Tables('users', 'orders', 'products', schema=schema)

# Create multiple basic columns
columns = Columns('id', 'name', 'email', 'created_at')

# More complex example
from pypika import MySQLQuery

# Tables using specific query class
user_table, order_table = Tables('users', 'orders', query_cls=MySQLQuery)

# Now these tables will use MySQL-specific features
query = user_table.select('*')  # Uses MySQLQuery internally

Temporal Table Support

Support for SQL temporal tables with FOR SYSTEM_TIME and FOR PORTION clauses.

class Table:
    def for_(self, temporal_criterion: Criterion) -> Table:
        """Add temporal FOR clause."""
        
    def for_portion(self, period_criterion: PeriodCriterion) -> Table:
        """Add FOR PORTION clause."""

Usage Examples:

from pypika import Table, SYSTEM_TIME
from pypika.functions import Now
from datetime import datetime

# Temporal table queries
users = Table('users')

# Query historical data
historical_users = users.for_(SYSTEM_TIME.between('2023-01-01', '2023-12-31'))
query = Query.from_(historical_users).select('*')

# Query as of specific time
as_of_users = users.for_(SYSTEM_TIME.as_of('2023-06-01'))
query = Query.from_(as_of_users).select('*')

# Period-based temporal queries
# (Requires period criterion implementation)
# users.for_portion(period_criterion)

Advanced Table Features

Additional table functionality including custom query class assignment and complex schema hierarchies.

Usage Examples:

from pypika import Table, Query, PostgreSQLQuery, MySQLQuery

# Table with custom query class
pg_users = Table('users', query_cls=PostgreSQLQuery)
mysql_users = Table('users', query_cls=MySQLQuery)

# PostgreSQL-specific features available
pg_query = pg_users.select('*')  # Can use PostgreSQL-specific methods

# Complex schema with different query classes
from pypika import Database, Schema

# Database with mixed query classes
postgres_db = Database('postgres_db')
mysql_db = Database('mysql_db')

# Tables inherit query capabilities based on their configuration
pg_users = Table('users', schema=postgres_db.public, query_cls=PostgreSQLQuery)
mysql_users = Table('users', schema=mysql_db.main, query_cls=MySQLQuery)

# Each table uses appropriate SQL dialect
pg_query = pg_users.select('*').limit(10)  # PostgreSQL LIMIT syntax
mysql_query = mysql_users.select('*').limit(10)  # MySQL LIMIT syntax

Install with Tessl CLI