repository·master·Indexed 25 days ago
https://github.com/kayak/pypikaPyPika is a Python query builder that provides a programmatic API for constructing SQL queries using the builder design pattern. It offers a flexible alternative to handwritten SQL or ORMs, supporting complex operations including joins, subqueries, set operations (UNION, INTERSECT, MINUS, EXCEPT), Common Table Expressions (WITH clauses), and window functions. It includes specialized support for MySQL and PostgreSQL insert constraint violations and allows for the implementation of custom SQL functions by extending the Function or AnalyticFunction classes.
Arithmetic operators (+, -, *, /) are implemented by pypika.Field. You can use these directly on Table attributes or Field instances. You can also alias the resulting expression using .as_().
Bitwise operations are supported via .bitwiseand() and .bitwiseor() methods on Field objects.
Use .groupby() for aggregation. Once a group is added, you can use .having() to filter aggregated results. For filtering based on window functions, use .qualify().
Aggregations can be performed using pypika.functions (aliased as fn).
Join tables using .join(). Joins must be immediately followed by either .on(criterion) or .using(*fields).
Supported join types include:
.left_join() / .left_outer_join().right_join() / .right_outer_join().inner_join().outer_join().full_outer_join().cross_join().hash_join()Use .on_field() as a shortcut to join when the field names are identical in both tables.
pre-commit to automate format checks. To install and configure these hooks in your local development environment, run the make install command.make installQuery.update(table) or by calling .update() directly on a Table instance. Use .set(column, value) to define updates. You can chain .where(), .join(), and .limit() to refine the update operation.pypika.Query class is the entry point for building SQL queries. For simple queries, you can use string names for tables and columns. For complex queries, use the pypika.Table class. To generate the raw SQL string, cast the query object to a str() or call .get_sql().Filter data using .where(). Multiple calls to .where() will append conditions using AND.
==, !=, >, etc.).BETWEEN (e.g., field[start:end]) and .isin([list]) for IN clauses.& (AND), | (OR), and ^ (XOR) to combine criteria.pypika.Criterion.all([list_of_criteria]) for AND chains and pypika.Criterion.any([list_of_criteria]) for OR chains.To prevent SQL injection and build prepared statements, use Parameter types. You must choose the type that matches your database driver's requirements (e.g., QmarkParameter for ?, NamedParameter for :name).
To extract both the SQL string and the parameter values for execution, pass a parameter object to .get_sql(parameter=...).
from pypika import Query, Table, QmarkParameter, NamedParameter
customers = Table('customers')
# 1. Using QmarkParameter (e.g., for SQLite)
q = Query.from_(customers).select('*').where((customers.status == 'active') & (customers.age >= 18))
parameter = QmarkParameter()
sql = q.get_sql(parameter=parameter)
params = parameter.get_parameters()
# sql: SELECT * FROM "customers" WHERE "status"=? AND "age"=?
# params: ['active', 18]
# 2. Using NamedParameter (e.g., for Vertica/Oracle)
q = Query.from_(customers).select('*').where(customers.status == 'active')
parameter = NamedParameter()
sql = q.get_sql(parameter=parameter)
params = parameter.get_parameters()
# sql: SELECT * FROM "customers" WHERE "status"=:param1
# params: {'param1': 'active'}The .pipe() method on QueryBuilder provides a readable way to chain custom functions that modify a query. This is an alternative to deeply nested function calls.
from pypika import Field, Query, functions as fn
from pypika.queries import QueryBuilder
def filter_days(query: QueryBuilder, col, num_days: int) -> QueryBuilder:
if isinstance(col, str):
col = Field(col)
return query.where(col > fn.Now() - num_days)
base_query = Query.from_("table")
query = (
base_query
.pipe(filter_days, "date", num_days=7)
)tox using the unittest framework. You can run the full test suite locally using the make test command. Note that these tests are also automatically executed by GitHub Actions on every pull request.make testYou can perform joins using .join() combined with .on_field() for specific column mappings or .using() for joins where the column name is identical in both tables.
Join with specific fields:
Use .on_field(field1, field2) to specify the join condition.
Join using USING clause:
Use .using(field_name) to generate a JOIN ... USING SQL statement.
history, customers = Tables('history', 'customers')
# Join with ON clause
q = Query \
.from_(history) \
.join(customers) \
.on_field('customer_id', 'group') \
.select(history.star) \
.where(customers.group == 'A')
# Join with USING clause
q = Query \
.from_(history) \
.join(customers) \
.using('customer_id') \
.select(history.star) \
.where(customers.id == 5)