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

SQL Functions

Name: tessl/pypi-pypika
Author: tessl

Comprehensive collection of SQL functions including aggregate functions (COUNT, SUM, AVG), string functions (CONCAT, SUBSTRING), date/time functions (NOW, DATE_ADD), mathematical functions (ABS, SQRT), and type conversion functions. These functions provide database-agnostic SQL functionality with dialect-specific optimizations.

Capabilities

Aggregate Functions

SQL aggregate functions for data summarization and statistical calculations.

class Count(AggregateFunction):
    def __init__(self, term, alias=None):
        """COUNT function with DISTINCT support."""
        
    def distinct(self) -> Count:
        """Add DISTINCT modifier."""

class Sum(AggregateFunction):
    def __init__(self, term, alias=None):
        """SUM function with DISTINCT support."""
        
    def distinct(self) -> Sum:
        """Add DISTINCT modifier."""

class Avg(AggregateFunction):
    def __init__(self, term, alias=None):
        """AVG function."""

class Min(AggregateFunction):
    def __init__(self, term, alias=None):
        """MIN function."""

class Max(AggregateFunction):
    def __init__(self, term, alias=None):
        """MAX function."""

class Std(AggregateFunction):
    def __init__(self, term, alias=None):
        """Standard deviation function."""

class StdDev(AggregateFunction):
    def __init__(self, term, alias=None):
        """STDDEV function."""

class First(AggregateFunction):
    def __init__(self, term, alias=None):
        """FIRST function."""

class Last(AggregateFunction):
    def __init__(self, term, alias=None):
        """LAST function."""

Usage Examples:

from pypika import Query, Table
from pypika.functions import Count, Sum, Avg, Min, Max, StdDev

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

# Basic aggregate functions
query = (Query.from_(orders)
         .select(
             Count('*').as_('total_orders'),
             Sum(orders.amount).as_('total_revenue'),
             Avg(orders.amount).as_('avg_order_value'),
             Min(orders.amount).as_('min_order'),
             Max(orders.amount).as_('max_order')
         ))

# COUNT with DISTINCT
query = (Query.from_(orders)
         .select(
             Count(orders.customer_id).distinct().as_('unique_customers'),
             Count('*').as_('total_orders')
         ))

# SUM with DISTINCT
query = (Query.from_(orders)
         .select(
             Sum(orders.amount).distinct().as_('unique_order_amounts')
         ))

# Grouped aggregations
query = (Query.from_(orders)
         .select(
             orders.customer_id,
             Count('*').as_('order_count'),
             Sum(orders.amount).as_('total_spent'),
             Avg(orders.amount).as_('avg_order_value'),
             StdDev(orders.amount).as_('order_amount_stddev')
         )
         .groupby(orders.customer_id))

# HAVING with aggregates
query = (Query.from_(orders)
         .select(
             orders.customer_id,
             Count('*').as_('order_count'),
             Sum(orders.amount).as_('total_spent')
         )
         .groupby(orders.customer_id)
         .having(Count('*') > 5)
         .having(Sum(orders.amount) > 1000))

String Functions

String manipulation and text processing functions.

class Concat(Function):
    def __init__(self, *terms):
        """Concatenate strings."""

class Upper(Function):
    def __init__(self, term):
        """Convert to uppercase."""

class Lower(Function):
    def __init__(self, term):
        """Convert to lowercase."""

class Length(Function):
    def __init__(self, term):
        """Get string length."""

class Substring(Function):
    def __init__(self, term, start, length=None):
        """Extract substring."""

class Trim(Function):
    def __init__(self, term, chars=None):
        """Trim whitespace or specified characters."""

class Replace(Function):
    def __init__(self, term, search, replacement):
        """Replace text in string."""

class Reverse(Function):
    def __init__(self, term):
        """Reverse string."""

class Ascii(Function):
    def __init__(self, term):
        """Get ASCII value of first character."""

class Insert(Function):
    def __init__(self, term, position, length, new_string):
        """Insert string at position."""

class SplitPart(Function):
    def __init__(self, term, delimiter, field_number):
        """Split string and return part."""

class RegexpMatches(Function):
    def __init__(self, term, pattern, flags=None):
        """Regular expression matches."""

class RegexpLike(Function):
    def __init__(self, term, pattern, flags=None):
        """Regular expression like."""

Usage Examples:

from pypika import Query, Table
from pypika.functions import (
    Concat, Upper, Lower, Length, Substring, Trim, Replace, 
    Reverse, SplitPart, RegexpLike
)

users = Table('users')
products = Table('products')

# String concatenation
query = (Query.from_(users)
         .select(
             Concat(users.first_name, ' ', users.last_name).as_('full_name'),
             Concat('User: ', users.username).as_('display_name')
         ))

# Case conversion
query = (Query.from_(users)
         .select(
             Upper(users.first_name).as_('first_name_upper'),
             Lower(users.email).as_('email_lower')
         ))

# String analysis
query = (Query.from_(products)
         .select(
             products.name,
             Length(products.name).as_('name_length'),
             Length(products.description).as_('desc_length')
         ))

# Substring extraction
query = (Query.from_(users)
         .select(
             users.email,
             Substring(users.email, 1, 3).as_('email_prefix'),
             Substring(users.phone, -4).as_('last_four_digits')
         ))

# String cleaning
query = (Query.from_(users)
         .select(
             Trim(users.first_name).as_('trimmed_name'),
             Replace(users.phone, '-', '').as_('phone_digits_only'),
             Replace(users.email, '@gmail.com', '@company.com').as_('company_email')
         ))

# String manipulation
query = (Query.from_(products)
         .select(
             products.code,
             Reverse(products.code).as_('reversed_code')
         ))

# String splitting and pattern matching
query = (Query.from_(users)
         .select(
             users.email,
             SplitPart(users.email, '@', 1).as_('username'),
             SplitPart(users.email, '@', 2).as_('domain')
         )
         .where(RegexpLike(users.email, r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$')))

Date and Time Functions

Date and time manipulation functions for temporal data processing.

class Now(Function):
    def __init__(self):
        """Current timestamp."""

class CurDate(Function):
    def __init__(self):
        """Current date."""

class CurTime(Function):
    def __init__(self):
        """Current time."""

class Date(Function):
    def __init__(self, term):
        """Extract date part."""

class Timestamp(Function):
    def __init__(self, term):
        """Convert to timestamp."""

class DateAdd(Function):
    def __init__(self, date_part, interval, date_term):
        """Add interval to date."""

class DateDiff(Function):
    def __init__(self, date_part, start_date, end_date):
        """Calculate date difference."""

class TimeDiff(Function):
    def __init__(self, start_time, end_time):
        """Calculate time difference."""

class Extract(Function):
    def __init__(self, date_part, term):
        """Extract date part."""

class ToDate(Function):
    def __init__(self, term, format_mask=None):
        """Convert to date."""

class TimestampAdd(Function):
    def __init__(self, unit, interval, timestamp):
        """Add interval to timestamp."""

class UtcTimestamp(Function):
    def __init__(self):
        """Current UTC timestamp."""

class CurTimestamp(Function):
    def __init__(self):
        """Current timestamp."""

Usage Examples:

from pypika import Query, Table, DatePart
from pypika.functions import (
    Now, CurDate, CurTime, Date, DateAdd, DateDiff, Extract, 
    ToDate, TimestampAdd, UtcTimestamp
)

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

# Current date/time functions
query = (Query.from_(users)
         .select(
             users.name,
             Now().as_('current_timestamp'),
             CurDate().as_('current_date'),
             CurTime().as_('current_time'),
             UtcTimestamp().as_('utc_timestamp')
         ))

# Date extraction
query = (Query.from_(orders)
         .select(
             orders.id,
             Date(orders.created_at).as_('order_date'),
             Extract(DatePart.year, orders.created_at).as_('order_year'),
             Extract(DatePart.month, orders.created_at).as_('order_month'),
             Extract(DatePart.day, orders.created_at).as_('order_day')
         ))

# Date arithmetic
query = (Query.from_(users)
         .select(
             users.name,
             users.created_at,
             DateAdd(DatePart.day, 30, users.created_at).as_('plus_30_days'),
             DateAdd(DatePart.month, 1, users.created_at).as_('plus_1_month'),
             DateAdd(DatePart.year, -1, Now()).as_('one_year_ago')
         ))

# Date differences
query = (Query.from_(users)
         .select(
             users.name,
             users.created_at,
             DateDiff(DatePart.day, users.created_at, Now()).as_('account_age_days'),
             DateDiff(DatePart.month, users.created_at, Now()).as_('account_age_months')
         ))

# Time-based filtering
thirty_days_ago = DateAdd(DatePart.day, -30, Now())
query = (Query.from_(orders)
         .select('*')
         .where(orders.created_at > thirty_days_ago))

# Grouping by date parts
query = (Query.from_(orders)
         .select(
             Extract(DatePart.year, orders.created_at).as_('year'),
             Extract(DatePart.month, orders.created_at).as_('month'),
             Count('*').as_('order_count'),
             Sum(orders.amount).as_('monthly_revenue')
         )
         .groupby(
             Extract(DatePart.year, orders.created_at),
             Extract(DatePart.month, orders.created_at)
         )
         .orderby('year', 'month'))

# String to date conversion
query = (Query.from_(events)
         .select(
             events.id,
             ToDate(events.date_string, 'YYYY-MM-DD').as_('parsed_date')
         )
         .where(ToDate(events.date_string, 'YYYY-MM-DD') > '2023-01-01'))

Mathematical Functions

Mathematical and arithmetic functions for numerical calculations.

class Abs(Function):
    def __init__(self, term):
        """Absolute value."""

class Sqrt(Function):
    def __init__(self, term):
        """Square root."""

class Floor(Function):
    def __init__(self, term):
        """Floor function."""

class Ceil(Function):
    def __init__(self, term):
        """Ceiling function."""

class Round(Function):
    def __init__(self, term, precision=None):
        """Round to specified precision."""

class Power(Function):
    def __init__(self, base, exponent):
        """Power function."""

class Mod(Function):
    def __init__(self, dividend, divisor):
        """Modulo operation."""

class ApproximatePercentile(Function):
    def __init__(self, term, percentile):
        """Approximate percentile calculation."""

Usage Examples:

from pypika import Query, Table
from pypika.functions import Abs, Sqrt, Floor, Round, Power, Mod

orders = Table('orders')
products = Table('products')
metrics = Table('metrics')

# Basic math functions
query = (Query.from_(orders)
         .select(
             orders.amount,
             Abs(orders.discount).as_('abs_discount'),
             Sqrt(orders.quantity).as_('sqrt_quantity'),
             Floor(orders.amount).as_('amount_floor'),
             Round(orders.amount, 2).as_('amount_rounded')
         ))

# Advanced calculations
query = (Query.from_(products)
         .select(
             products.name,
             products.price,
             Power(products.price, 2).as_('price_squared'),
             Mod(products.id, 10).as_('id_mod_10')
         ))

# Financial calculations
query = (Query.from_(orders)
         .select(
             orders.customer_id,
             Round(Avg(orders.amount), 2).as_('avg_order_value'),
             Round(Sqrt(Sum(Power(orders.amount - Avg(orders.amount), 2)) / Count('*')), 2).as_('std_dev')
         )
         .groupby(orders.customer_id))

Type Conversion Functions

Functions for converting between different data types.

class Cast(Function):
    def __init__(self, term, as_type):
        """CAST function for type conversion."""

class Convert(Function):
    def __init__(self, term, as_type):
        """CONVERT function for type conversion."""

class ToChar(Function):
    def __init__(self, term, format_mask=None):
        """Convert to character string."""

class Signed(Function):
    def __init__(self, term):
        """Convert to signed integer."""

class Unsigned(Function):
    def __init__(self, term):
        """Convert to unsigned integer."""

Usage Examples:

from pypika import Query, Table
from pypika.functions import Cast, Convert, ToChar, Signed

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

# Type casting
query = (Query.from_(users)
         .select(
             users.id,
             Cast(users.id, 'VARCHAR').as_('id_string'),
             Cast(users.created_at, 'DATE').as_('created_date'),
             Cast(users.score, 'INTEGER').as_('score_int')
         ))

# String formatting
query = (Query.from_(orders)
         .select(
             orders.id,
             ToChar(orders.created_at, 'YYYY-MM-DD').as_('formatted_date'),
             ToChar(orders.amount, '999,999.99').as_('formatted_amount')
         ))

# Numeric conversions
query = (Query.from_(users)
         .select(
             users.id,
             Signed(users.score).as_('signed_score')
         ))

Null Handling Functions

Functions for handling NULL values and providing default values.

class IsNull(Function):
    def __init__(self, term):
        """Check if value is NULL."""

class Coalesce(Function):
    def __init__(self, *terms):
        """Return first non-NULL value."""

class IfNull(Function):
    def __init__(self, term, replacement):
        """Replace NULL with specified value."""

class NullIf(Function):
    def __init__(self, term1, term2):
        """Return NULL if values are equal."""

class NVL(Function):
    def __init__(self, term, replacement):
        """Oracle-style NULL value replacement."""

Usage Examples:

from pypika import Query, Table, NULL
from pypika.functions import Coalesce, IfNull, NullIf

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

# NULL handling
query = (Query.from_(users)
         .select(
             users.id,
             Coalesce(users.nickname, users.first_name, 'Anonymous').as_('display_name'),
             IfNull(users.phone, 'No phone').as_('phone_display'),
             NullIf(users.middle_name, '').as_('middle_name_cleaned')
         ))

# Default values for aggregates
query = (Query.from_(orders)
         .select(
             orders.customer_id,
             Coalesce(Sum(orders.amount), 0).as_('total_spent'),
             Coalesce(Count(orders.id), 0).as_('order_count')
         )
         .groupby(orders.customer_id))

# Conditional null replacement
query = (Query.from_(users)
         .select(
             users.name,
             Coalesce(
                 NullIf(users.preferred_email, ''),
                 users.email,
                 'no-email@company.com'
             ).as_('contact_email')
         ))

Utility Functions

Additional utility functions for specialized operations.

class Bin(Function):
    def __init__(self, term):
        """Convert to binary representation."""

class Ascii(Function):
    def __init__(self, term):
        """Get ASCII value."""

class Insert(Function):
    def __init__(self, term, position, length, new_string):
        """Insert string at position."""

Usage Examples:

from pypika import Query, Table
from pypika.functions import Bin, Ascii

users = Table('users')

# Utility functions
query = (Query.from_(users)
         .select(
             users.id,
             Bin(users.id).as_('id_binary'),
             Ascii(users.first_name).as_('first_char_ascii')
         ))

Install with Tessl CLI