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

Terms and Expressions

Name: tessl/pypi-pypika
Author: tessl

Building blocks for SQL expressions including fields, criteria, case statements, arrays, tuples, and custom functions. These components support all SQL operations and can be combined to create complex expressions with full operator support and type safety.

Capabilities

Field Operations

Field references supporting all SQL operations, comparisons, and transformations.

class Field:
    def __init__(self, name: str, alias: Optional[str] = None, table: Optional[Table] = None):
        """
        Create field reference.
        
        Parameters:
        - name: Field name
        - alias: Field alias
        - table: Parent table
        """
        
    def as_(self, alias: str) -> Field:
        """Set field alias."""
        
    def asc(self) -> Order:
        """Create ascending order."""
        
    def desc(self) -> Order:
        """Create descending order."""
        
    def isnull(self) -> BasicCriterion:
        """Check if field is NULL."""
        
    def notnull(self) -> BasicCriterion:
        """Check if field is NOT NULL."""
        
    def isin(self, *values) -> BasicCriterion:
        """Check if field IN values."""
        
    def notin(self, *values) -> BasicCriterion:
        """Check if field NOT IN values."""
        
    def between(self, lower, upper) -> RangeCriterion:
        """Check if field BETWEEN lower and upper."""
        
    def like(self, pattern) -> BasicCriterion:
        """Pattern matching with LIKE."""
        
    def not_like(self, pattern) -> BasicCriterion:
        """Pattern matching with NOT LIKE."""
        
    def rlike(self, pattern) -> BasicCriterion:
        """Regular expression matching with RLIKE."""
        
    def not_rlike(self, pattern) -> BasicCriterion:
        """Regular expression matching with NOT RLIKE."""
        
    def distinct(self) -> DistinctOptionField:
        """Mark field as DISTINCT."""
        
    # Arithmetic operations
    def __add__(self, other):
        """Addition: field + other"""
        
    def __sub__(self, other):
        """Subtraction: field - other"""
        
    def __mul__(self, other):
        """Multiplication: field * other"""
        
    def __truediv__(self, other):
        """Division: field / other"""
        
    def __mod__(self, other):
        """Modulo: field % other"""
        
    # Comparison operations
    def __eq__(self, other):
        """Equality: field == other"""
        
    def __ne__(self, other):
        """Inequality: field != other"""
        
    def __lt__(self, other):
        """Less than: field < other"""
        
    def __le__(self, other):
        """Less than or equal: field <= other"""
        
    def __gt__(self, other):
        """Greater than: field > other"""
        
    def __ge__(self, other):
        """Greater than or equal: field >= other"""

Usage Examples:

from pypika import Table, Query, Field

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

# Field comparisons
adult_users = users.age >= 18
active_users = users.status == 'active'
recent_orders = orders.created_at > '2023-01-01'

# Null checks
users_with_email = users.email.notnull()
users_without_phone = users.phone.isnull()

# IN operations
vip_statuses = users.status.isin('gold', 'platinum', 'diamond')
excluded_ids = users.id.notin(1, 2, 3)

# Pattern matching
gmail_users = users.email.like('%@gmail.com')
non_temp_users = users.email.not_like('temp_%')

# BETWEEN operations
middle_aged = users.age.between(25, 65)
recent_range = orders.created_at.between('2023-01-01', '2023-12-31')

# Arithmetic operations
total_with_tax = orders.subtotal + orders.tax
discount_percent = (orders.discount / orders.subtotal) * 100
monthly_salary = users.annual_salary / 12

# Field aliasing
full_name = (users.first_name + ' ' + users.last_name).as_('full_name')
age_group = Field(
    Case()
    .when(users.age < 18, 'Minor')
    .when(users.age < 65, 'Adult')
    .else_('Senior')
).as_('age_group')

# Ordering
query = Query.from_(users).select('*').orderby(users.created_at.desc(), users.name.asc())

# DISTINCT fields
query = Query.from_(users).select(users.status.distinct())

Criterion Operations

Boolean expressions for WHERE clauses with logical operations and complex condition building.

class Criterion:
    @staticmethod
    def any(*terms) -> EmptyCriterion:
        """Create OR-combined criterion from multiple terms."""
        
    @staticmethod
    def all(*terms) -> EmptyCriterion:
        """Create AND-combined criterion from multiple terms."""
        
    # Logical operations
    def __and__(self, other):
        """Logical AND: criterion & other"""
        
    def __or__(self, other):
        """Logical OR: criterion | other"""
        
    def __invert__(self):
        """Logical NOT: ~criterion"""

class EmptyCriterion(Criterion):
    """Neutral element for criterion combinations."""

Usage Examples:

from pypika import Table, Query, Criterion

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

# Simple conditions
adult_criterion = users.age >= 18
active_criterion = users.status == 'active'

# Combining conditions with logical operators
adult_active = adult_criterion & active_criterion
inactive_or_young = ~active_criterion | (users.age < 18)

# Complex condition building
premium_users = (
    (users.status.isin('gold', 'platinum')) &
    (users.total_orders > 10) &
    (users.last_login > '2023-01-01')
)

# Using Criterion.all() and Criterion.any()
all_conditions = Criterion.all([
    users.age >= 18,
    users.status == 'active',
    users.email.notnull()
])

any_conditions = Criterion.any([
    users.status == 'premium',
    users.total_spent > 1000,
    users.referral_count > 5
])

# Nested conditions
complex_criterion = (
    Criterion.all([
        users.status == 'active',
        users.age >= 18
    ]) |
    Criterion.any([
        users.is_vip == True,
        users.total_spent > 5000
    ])
)

# Using in queries
query = Query.from_(users).select('*').where(complex_criterion)

Case Expressions

SQL CASE statement builder for conditional logic and value transformation.

class Case:
    def when(self, criterion: Criterion, term) -> Case:
        """
        Add WHEN clause.
        
        Parameters:
        - criterion: Condition to check
        - term: Value to return if condition is true
        """
        
    def else_(self, term) -> Case:
        """
        Add ELSE clause.
        
        Parameters:
        - term: Default value to return
        """

Usage Examples:

from pypika import Table, Query, Case

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

# Simple CASE expression
status_description = (
    Case()
    .when(users.status == 'active', 'User is active')
    .when(users.status == 'inactive', 'User is inactive')
    .else_('Unknown status')
).as_('status_description')

# Numeric CASE
age_category = (
    Case()
    .when(users.age < 18, 'Minor')
    .when(users.age < 30, 'Young Adult')
    .when(users.age < 50, 'Adult')
    .when(users.age < 65, 'Middle Aged')
    .else_('Senior')
).as_('age_category')

# Complex conditions in CASE
user_tier = (
    Case()
    .when((users.total_spent > 10000) & (users.years_active > 5), 'Platinum')
    .when(users.total_spent > 5000, 'Gold')
    .when(users.total_spent > 1000, 'Silver')
    .else_('Bronze')
).as_('user_tier')

# CASE in SELECT
query = Query.from_(users).select(
    users.name,
    users.email,
    status_description,
    age_category,
    user_tier
)

# CASE in WHERE clause
query = Query.from_(users).select('*').where(
    Case()
    .when(users.role == 'admin', users.last_login > '2023-01-01')
    .else_(users.last_login > '2023-06-01')
)

# CASE in ORDER BY
query = Query.from_(users).select('*').orderby(
    Case()
    .when(users.status == 'active', 1)
    .when(users.status == 'pending', 2)
    .else_(3)
)

Array and Tuple Operations

SQL array and tuple literal support for complex data structures.

class Array:
    def __init__(self, *items):
        """
        Create SQL array literal.
        
        Parameters:
        - items: Array elements
        """

class Tuple:
    def __init__(self, *items):
        """
        Create SQL tuple literal.
        
        Parameters:
        - items: Tuple elements
        """

Usage Examples:

from pypika import Table, Query, Array, Tuple

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

# Array operations
user_roles = Array('admin', 'user', 'moderator')
status_array = Array('active', 'pending', 'inactive')

# Using arrays in queries
query = Query.from_(users).select('*').where(
    users.role.isin(user_roles)
)

# Tuple operations for multi-column conditions
user_tuple = Tuple(users.first_name, users.last_name)
name_combinations = Tuple(
    Tuple('John', 'Doe'),
    Tuple('Jane', 'Smith'),
    Tuple('Bob', 'Johnson')
)

query = Query.from_(users).select('*').where(
    user_tuple.isin(name_combinations)
)

# Arrays in PostgreSQL-specific operations (when using PostgreSQLQuery)
from pypika import PostgreSQLQuery
pg_users = Table('users', query_cls=PostgreSQLQuery)

# Array contains operations (PostgreSQL specific)
tags_array = Array('python', 'sql', 'data')
# Note: Specific array operations depend on dialect support

JSON Operations

JSON field operations and path-based access for PostgreSQL-style JSON handling.

class JSON:
    def get_json_value(self, key_or_index) -> BasicCriterion:
        """JSON -> operator: extract JSON value."""
        
    def get_text_value(self, key_or_index) -> BasicCriterion:
        """JSON ->> operator: extract JSON value as text."""
        
    def get_path_json_value(self, path_json) -> BasicCriterion:
        """JSON #> operator: extract JSON value at path."""
        
    def get_path_text_value(self, path_json) -> BasicCriterion:
        """JSON #>> operator: extract JSON value at path as text."""
        
    def has_key(self, key) -> BasicCriterion:
        """JSON ? operator: check if key exists."""
        
    def contains(self, other) -> BasicCriterion:
        """JSON @> operator: check if contains value."""
        
    def contained_by(self, other) -> BasicCriterion:
        """JSON <@ operator: check if contained by value."""
        
    def has_keys(self, keys) -> BasicCriterion:
        """JSON ?& operator: check if has all keys."""
        
    def has_any_keys(self, keys) -> BasicCriterion:
        """JSON ?| operator: check if has any keys."""

Usage Examples:

from pypika import Table, Query, JSON, PostgreSQLQuery

# Table with JSON fields (typically PostgreSQL)
users = Table('users', query_cls=PostgreSQLQuery)

# JSON field operations
profile = users.profile  # Assuming 'profile' is a JSON field

# Extract JSON values
name = profile.get_json_value('name')
age = profile.get_text_value('age')

# Path-based extraction
address_city = profile.get_path_text_value(['address', 'city'])
phone_numbers = profile.get_path_json_value(['contact', 'phones'])

# Key existence checks
has_email = profile.has_key('email')
has_address = profile.has_key('address')

# Multiple key checks
has_contact_info = profile.has_keys(['email', 'phone'])
has_any_social = profile.has_any_keys(['twitter', 'facebook', 'linkedin'])

# Contains operations
has_premium = profile.contains({'subscription': 'premium'})
is_admin = profile.contained_by({'roles': ['admin', 'user']})

# Using in queries
query = Query.from_(users).select(
    users.id,
    profile.get_text_value('name').as_('profile_name'),
    profile.get_path_text_value(['address', 'city']).as_('city')
).where(
    profile.has_key('email') & 
    profile.get_text_value('status').eq('active')
)

Custom Functions

Factory for creating user-defined SQL functions with flexible parameter handling.

class CustomFunction:
    def __init__(self, name: str, params: Optional[list] = None):
        """
        Create custom SQL function.
        
        Parameters:
        - name: Function name
        - params: Parameter specifications
        """
        
    def __call__(self, *args) -> Function:
        """Call function with arguments."""

Usage Examples:

from pypika import Table, Query, CustomFunction

users = Table('users')

# Define custom functions
calculate_age = CustomFunction('CALCULATE_AGE')
hash_email = CustomFunction('HASH_EMAIL')
format_phone = CustomFunction('FORMAT_PHONE')
distance_between = CustomFunction('DISTANCE_BETWEEN')

# Use custom functions in queries
query = Query.from_(users).select(
    users.name,
    calculate_age(users.birth_date).as_('age'),
    hash_email(users.email).as_('email_hash'),
    format_phone(users.phone).as_('formatted_phone')
)

# Custom aggregation functions
custom_avg = CustomFunction('CUSTOM_AVG')
custom_count = CustomFunction('COUNT_DISTINCT_VALUES')

query = Query.from_(users).select(
    custom_avg(users.score).as_('avg_score'),
    custom_count(users.category).as_('unique_categories')
).groupby(users.department)

# Custom window functions
custom_rank = CustomFunction('CUSTOM_RANK')
custom_lag = CustomFunction('PREV_VALUE')

query = Query.from_(users).select(
    users.name,
    users.salary,
    custom_rank().over(users.department).orderby(users.salary.desc()).as_('dept_rank'),
    custom_lag(users.salary, 1).over(users.department).orderby(users.hire_date).as_('prev_salary')
)

# Geographic/spatial functions
st_distance = CustomFunction('ST_DISTANCE')
st_within = CustomFunction('ST_WITHIN')

locations = Table('locations')
query = Query.from_(locations).select(
    locations.name,
    st_distance(locations.point, 'POINT(0 0)').as_('distance_from_origin')
).where(
    st_within(locations.point, 'POLYGON((0 0, 1 0, 1 1, 0 1, 0 0))')
)

Special Values and Constants

Built-in constants and special value representations.

class NullValue:
    """SQL NULL value."""

class SystemTimeValue:
    """SQL SYSTEM_TIME value for temporal tables."""

class Not:
    def __init__(self, term):
        """NOT operator wrapper."""

class Bracket:
    def __init__(self, term):
        """Parentheses wrapper for expressions."""

class Index:
    def __init__(self, name: str):
        """Database index reference."""

Usage Examples:

from pypika import Table, Query, NULL, SYSTEM_TIME, Not, Bracket, Index

users = Table('users')

# NULL operations
query = Query.from_(users).select('*').where(users.deleted_at == NULL)
query = Query.from_(users).select('*').where(users.phone != NULL)

# NOT operations
not_admin = Not(users.role == 'admin')
query = Query.from_(users).select('*').where(not_admin)

# Explicit parentheses
complex_condition = Bracket(
    (users.age > 18) & (users.status == 'active')
) | Bracket(
    (users.role == 'admin') & (users.verified == True)
)

query = Query.from_(users).select('*').where(complex_condition)

# Index hints (MySQL specific)
user_email_idx = Index('idx_user_email')
# Usage depends on dialect-specific implementation

# SYSTEM_TIME for temporal tables
historical_query = Query.from_(users.for_(SYSTEM_TIME.between('2023-01-01', '2023-12-31')))

Interval Operations

Time and date interval expressions for temporal calculations.

class Interval:
    def __init__(self, years=0, months=0, days=0, hours=0, minutes=0, 
                 seconds=0, microseconds=0, quarters=0, weeks=0, dialect=None):
        """
        Create time interval.
        
        Parameters:
        - years, months, days, hours, minutes, seconds, microseconds: Time components
        - quarters, weeks: Additional time units
        - dialect: Database dialect for interval formatting
        """

Usage Examples:

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

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

# Date arithmetic with intervals
one_year_ago = Now() - Interval(years=1)
thirty_days_ago = Now() - Interval(days=30)
six_months = Interval(months=6)

# Recent users (last 30 days)
query = Query.from_(users).select('*').where(
    users.created_at > thirty_days_ago
)

# Users created more than a year ago
query = Query.from_(users).select('*').where(
    users.created_at < one_year_ago
)

# Calculate future dates
renewal_date = users.subscription_start + Interval(years=1)
trial_end = users.trial_start + Interval(days=14)

query = Query.from_(users).select(
    users.name,
    users.subscription_start,
    renewal_date.as_('renewal_date'),
    trial_end.as_('trial_end')
)

# Complex interval calculations
quarterly_review = users.hire_date + Interval(quarters=1)
annual_bonus_date = users.hire_date + Interval(years=1, days=-1)

# Filter by interval ranges
recent_orders = Query.from_(orders).select('*').where(
    orders.created_at.between(
        Now() - Interval(days=7),
        Now()
    )
)

Install with Tessl CLI