tessl/pypi-sqlalchemy
Python SQL toolkit and Object Relational Mapper providing full power and flexibility of SQL
sql-expression.mddocs/
SQL Expression Language
Programmatic SQL construction with Python expressions, including SELECT, INSERT, UPDATE, DELETE statements, joins, subqueries, functions, and complex query composition. The SQL Expression Language provides database-agnostic SQL generation.
Capabilities
Query Construction Functions
Core functions for building SQL statements programmatically.
def select(*columns):
"""
Create SELECT statement.
Parameters:
- columns: Table, Column, or expression objects to select
Returns:
Select: SELECT statement object
"""
def insert(table):
"""
Create INSERT statement for table.
Parameters:
- table: Table object to insert into
Returns:
Insert: INSERT statement object
"""
def update(table):
"""
Create UPDATE statement for table.
Parameters:
- table: Table object to update
Returns:
Update: UPDATE statement object
"""
def delete(table):
"""
Create DELETE statement for table.
Parameters:
- table: Table object to delete from
Returns:
Delete: DELETE statement object
"""
def text(text, bind=None):
"""
Create literal SQL text with parameter binding.
Parameters:
- text: str, SQL text with optional parameter placeholders
- bind: Engine or Connection, bind for execution
Returns:
TextClause: Literal SQL text object
"""SELECT Statement Construction
Building and customizing SELECT queries with joins, filtering, grouping, and ordering.
class Select:
"""SELECT statement with query building methods."""
def where(self, *criteria):
"""
Add WHERE criteria to SELECT.
Parameters:
- criteria: Column expressions for filtering
Returns:
Select: Modified SELECT with WHERE clause
"""
def join(self, target, onclause=None, isouter=False, full=False):
"""
Add JOIN to SELECT.
Parameters:
- target: Table or selectable to join
- onclause: join condition (auto-detected if None)
- isouter: bool, use LEFT OUTER JOIN
- full: bool, use FULL OUTER JOIN
Returns:
Select: Modified SELECT with JOIN
"""
def outerjoin(self, target, onclause=None, full=False):
"""
Add LEFT OUTER JOIN to SELECT.
Parameters:
- target: Table or selectable to join
- onclause: join condition
- full: bool, use FULL OUTER JOIN
Returns:
Select: Modified SELECT with OUTER JOIN
"""
def order_by(self, *clauses):
"""
Add ORDER BY to SELECT.
Parameters:
- clauses: Column expressions or ordering functions
Returns:
Select: Modified SELECT with ORDER BY
"""
def group_by(self, *clauses):
"""
Add GROUP BY to SELECT.
Parameters:
- clauses: Column expressions for grouping
Returns:
Select: Modified SELECT with GROUP BY
"""
def having(self, *criteria):
"""
Add HAVING criteria to SELECT.
Parameters:
- criteria: Column expressions for HAVING clause
Returns:
Select: Modified SELECT with HAVING
"""
def limit(self, limit):
"""
Add LIMIT to SELECT.
Parameters:
- limit: int, maximum number of rows
Returns:
Select: Modified SELECT with LIMIT
"""
def offset(self, offset):
"""
Add OFFSET to SELECT.
Parameters:
- offset: int, number of rows to skip
Returns:
Select: Modified SELECT with OFFSET
"""
def distinct(self, *expr):
"""
Make SELECT DISTINCT.
Parameters:
- expr: optional expressions for DISTINCT ON (PostgreSQL)
Returns:
Select: Modified SELECT with DISTINCT
"""INSERT, UPDATE, DELETE Statements
Data modification statement construction with value binding and conditional operations.
class Insert:
"""INSERT statement with value specification."""
def values(self, *args, **kwargs):
"""
Specify values for INSERT.
Parameters:
- args: dictionaries of column-value pairs
- kwargs: column-value pairs as keyword arguments
Returns:
Insert: Modified INSERT with VALUES
"""
def returning(self, *cols):
"""
Add RETURNING clause (PostgreSQL, SQL Server, Oracle).
Parameters:
- cols: columns to return after insert
Returns:
Insert: Modified INSERT with RETURNING
"""
class Update:
"""UPDATE statement with WHERE and SET clauses."""
def where(self, *criteria):
"""
Add WHERE criteria to UPDATE.
Parameters:
- criteria: Column expressions for filtering
Returns:
Update: Modified UPDATE with WHERE
"""
def values(self, *args, **kwargs):
"""
Specify SET values for UPDATE.
Parameters:
- args: dictionaries of column-value pairs
- kwargs: column-value pairs as keyword arguments
Returns:
Update: Modified UPDATE with SET values
"""
def returning(self, *cols):
"""
Add RETURNING clause.
Parameters:
- cols: columns to return after update
Returns:
Update: Modified UPDATE with RETURNING
"""
class Delete:
"""DELETE statement with WHERE clause."""
def where(self, *criteria):
"""
Add WHERE criteria to DELETE.
Parameters:
- criteria: Column expressions for filtering
Returns:
Delete: Modified DELETE with WHERE
"""
def returning(self, *cols):
"""
Add RETURNING clause.
Parameters:
- cols: columns to return after delete
Returns:
Delete: Modified DELETE with RETURNING
"""Boolean and Logical Operators
Combining conditions with logical operators for complex WHERE clauses.
def and_(*clauses):
"""
Combine expressions with AND.
Parameters:
- clauses: Boolean expressions to AND together
Returns:
BooleanClauseList: Combined AND expression
"""
def or_(*clauses):
"""
Combine expressions with OR.
Parameters:
- clauses: Boolean expressions to OR together
Returns:
BooleanClauseList: Combined OR expression
"""
def not_(clause):
"""
Negate expression with NOT.
Parameters:
- clause: Boolean expression to negate
Returns:
UnaryExpression: Negated expression
"""Functions and Expressions
SQL functions, type casting, and conditional expressions.
def case(*whens, **kw):
"""
Create CASE expression.
Parameters:
- whens: sequence of (condition, value) tuples
- value: optional positional value for simple CASE
- else_: default value for ELSE clause
Returns:
Case: CASE expression
"""
def cast(expression, type_):
"""
Create CAST expression.
Parameters:
- expression: expression to cast
- type_: target data type
Returns:
Cast: CAST expression
"""
def extract(field, expr):
"""
Create EXTRACT expression for date/time components.
Parameters:
- field: str, component to extract (year, month, day, etc.)
- expr: date/time expression
Returns:
Extract: EXTRACT expression
"""
class func:
"""Namespace for SQL functions."""
@staticmethod
def count(expr=None):
"""
COUNT aggregate function.
Parameters:
- expr: expression to count (defaults to COUNT(*))
Returns:
Function: COUNT function call
"""
@staticmethod
def sum(expr):
"""
SUM aggregate function.
Parameters:
- expr: numeric expression to sum
Returns:
Function: SUM function call
"""
@staticmethod
def avg(expr):
"""
AVG aggregate function.
Parameters:
- expr: numeric expression to average
Returns:
Function: AVG function call
"""
@staticmethod
def max(expr):
"""
MAX aggregate function.
Parameters:
- expr: expression to find maximum
Returns:
Function: MAX function call
"""
@staticmethod
def min(expr):
"""
MIN aggregate function.
Parameters:
- expr: expression to find minimum
Returns:
Function: MIN function call
"""
@staticmethod
def now():
"""
Current timestamp function (database-specific).
Returns:
Function: Current timestamp function
"""
@staticmethod
def coalesce(*args):
"""
COALESCE function - return first non-null value.
Parameters:
- args: expressions to check for null
Returns:
Function: COALESCE function call
"""Subqueries and CTEs
Subquery construction and Common Table Expressions for complex queries.
def exists(element):
"""
Create EXISTS expression.
Parameters:
- element: SELECT statement for EXISTS check
Returns:
Exists: EXISTS expression
"""
def cte(selectable, name=None, recursive=False):
"""
Create Common Table Expression.
Parameters:
- selectable: SELECT statement for CTE
- name: str, CTE name (auto-generated if None)
- recursive: bool, create recursive CTE
Returns:
CTE: Common Table Expression
"""
class Subquery:
"""Subquery that can be used in FROM clauses."""
def as_(self, name):
"""
Create alias for subquery.
Parameters:
- name: str, alias name
Returns:
Alias: Aliased subquery
"""
class CTE:
"""Common Table Expression."""
def union(self, other):
"""
UNION with another CTE (for recursive CTEs).
Parameters:
- other: CTE or SELECT to union with
Returns:
CompoundSelect: UNION of CTEs
"""Set Operations
UNION, INTERSECT, and EXCEPT operations for combining query results.
def union(*selects):
"""
UNION multiple SELECT statements.
Parameters:
- selects: SELECT statements to union
Returns:
CompoundSelect: UNION query
"""
def union_all(*selects):
"""
UNION ALL multiple SELECT statements.
Parameters:
- selects: SELECT statements to union
Returns:
CompoundSelect: UNION ALL query
"""
def intersect(*selects):
"""
INTERSECT multiple SELECT statements.
Parameters:
- selects: SELECT statements to intersect
Returns:
CompoundSelect: INTERSECT query
"""
def except_(*selects):
"""
EXCEPT multiple SELECT statements.
Parameters:
- selects: SELECT statements for except operation
Returns:
CompoundSelect: EXCEPT query
"""Ordering and Grouping
Sort order specification and aggregation grouping.
def asc(column):
"""
Create ascending sort order.
Parameters:
- column: column expression to sort
Returns:
UnaryExpression: Ascending sort expression
"""
def desc(column):
"""
Create descending sort order.
Parameters:
- column: column expression to sort
Returns:
UnaryExpression: Descending sort expression
"""
def nulls_first(column):
"""
Sort nulls first.
Parameters:
- column: column expression with sort order
Returns:
UnaryExpression: Modified sort with nulls first
"""
def nulls_last(column):
"""
Sort nulls last.
Parameters:
- column: column expression with sort order
Returns:
UnaryExpression: Modified sort with nulls last
"""Window Functions
Window function support with OVER clauses.
def over(element, partition_by=None, order_by=None, rows=None, range_=None):
"""
Create OVER clause for window functions.
Parameters:
- element: function expression for windowing
- partition_by: expressions for PARTITION BY
- order_by: expressions for ORDER BY
- rows: tuple for ROWS frame specification
- range_: tuple for RANGE frame specification
Returns:
Over: Window function with OVER clause
"""
class Over:
"""Window function with OVER clause."""
def partition_by(self, *clauses):
"""
Set PARTITION BY for window.
Parameters:
- clauses: expressions for partitioning
Returns:
Over: Modified window with PARTITION BY
"""
def order_by(self, *clauses):
"""
Set ORDER BY for window.
Parameters:
- clauses: expressions for ordering
Returns:
Over: Modified window with ORDER BY
"""Literal Values and Parameters
Literal value binding and parameter specification.
def literal(value, type_=None):
"""
Create literal value expression.
Parameters:
- value: literal value
- type_: optional type specification
Returns:
Literal: Literal value expression
"""
def bindparam(key, value=None, type_=None):
"""
Create bound parameter placeholder.
Parameters:
- key: str, parameter name
- value: default parameter value
- type_: parameter data type
Returns:
BindParameter: Parameter placeholder
"""
def null():
"""
Create NULL literal.
Returns:
Null: NULL literal expression
"""
def true():
"""
Create TRUE literal.
Returns:
True_: TRUE literal expression
"""
def false():
"""
Create FALSE literal.
Returns:
False_: FALSE literal expression
"""Usage Examples
Basic SELECT Query
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, select
engine = create_engine("sqlite:///example.db")
metadata = MetaData()
users = Table('users', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(50)),
Column('email', String(100))
)
# Simple select
stmt = select(users).where(users.c.name.like('%John%'))
with engine.connect() as conn:
result = conn.execute(stmt)
rows = result.fetchall()Complex Query with Joins
from sqlalchemy import select, and_, or_, func
# Join with aggregation
stmt = select(
users.c.name,
func.count(orders.c.id).label('order_count')
).select_from(
users.join(orders, users.c.id == orders.c.user_id)
).where(
and_(
users.c.active == True,
orders.c.status.in_(['completed', 'shipped'])
)
).group_by(users.c.name).having(
func.count(orders.c.id) > 5
).order_by(desc(func.count(orders.c.id)))INSERT with RETURNING
from sqlalchemy import insert
stmt = insert(users).values(
name='New User',
email='new@example.com'
).returning(users.c.id)
with engine.connect() as conn:
result = conn.execute(stmt)
new_id = result.scalar()Subquery Example
# Subquery for filtering
subq = select(func.avg(users.c.age)).scalar_subquery()
stmt = select(users).where(users.c.age > subq)
# CTE example
users_cte = select(users.c.id, users.c.name).cte('users_cte')
stmt = select(users_cte.c.name).join(
orders, users_cte.c.id == orders.c.user_id
)Install with Tessl CLI
npx tessl i tessl/pypi-sqlalchemy