from __future__ import annotations import sys from collections.abc import Sequence from copy import copy from functools import reduce from typing import TYPE_CHECKING, Any, Generic, TypeVar from pypika.enums import Dialects, JoinType, ReferenceOption, SetOperation from pypika.terms import ( ArithmeticExpression, Criterion, EmptyCriterion, Field, Function, Index, Node, PeriodCriterion, Rollup, Star, Term, Tuple, ValueWrapper, ) from pypika.utils import ( JoinException, QueryException, RollupException, SetOperationException, builder, format_alias_sql, format_quotes, ignore_copy, ) if TYPE_CHECKING: if sys.version_info >= (3, 11): from typing import Self else: from typing_extensions import Self __author__ = "Timothy Heys" __email__ = "theys@kayak.com" QB = TypeVar("QB", bound="QueryBuilder") class Selectable(Node): def __init__(self, alias: str) -> None: self.alias = alias @builder def as_(self, alias: str) -> None: self.alias = alias def field(self, name: str) -> Field: return Field(name, table=self) @property def star(self) -> Star: return Star(self) @ignore_copy def __getattr__(self, name: str) -> Field: return self.field(name) @ignore_copy def __getitem__(self, name: str) -> Field: return self.field(name) def get_table_name(self) -> str: return self.alias class AliasedQuery(Selectable): def __init__(self, name: str, query: Selectable | None = None) -> None: super().__init__(alias=name) self.name = name self.query = query def get_sql(self, **kwargs: Any) -> str: if self.query is None: return self.name return self.query.get_sql(**kwargs) def __eq__(self, other: AliasedQuery) -> bool: return isinstance(other, AliasedQuery) and self.name == other.name def __hash__(self) -> int: return hash(str(self.name)) class Schema: def __init__(self, name: str, parent: Schema | None = None) -> None: self._name = name self._parent = parent def __eq__(self, other: Schema) -> bool: return isinstance(other, Schema) and self._name == other._name and self._parent == other._parent def __ne__(self, other: Schema) -> bool: return not self.__eq__(other) @ignore_copy def __getattr__(self, item: str) -> Table: return Table(item, schema=self) def get_sql(self, quote_char: str | None = None, **kwargs: Any) -> str: schema_sql = format_quotes(self._name, quote_char) if self._parent is not None: return "{parent}.{schema}".format( parent=self._parent.get_sql(quote_char=quote_char, **kwargs), schema=schema_sql, ) return schema_sql class Database(Schema): @ignore_copy def __getattr__(self, item: str) -> Schema: return Schema(item, parent=self) class Table(Selectable): @staticmethod def _init_schema(schema: str | list | tuple | Schema | None) -> str | list | tuple | Schema | None: # This is a bit complicated in order to support backwards compatibility. It should probably be cleaned up for # the next major release. Schema is accepted as a string, list/tuple, Schema instance, or None if isinstance(schema, Schema): return schema if isinstance(schema, (list, tuple)): return reduce(lambda obj, s: Schema(s, parent=obj), schema[1:], Schema(schema[0])) if schema is not None: return Schema(schema) return None def __init__( self, name: str, schema: Schema | str | None = None, alias: str | None = None, query_cls: type[Query] | None = None, ) -> None: super().__init__(alias) self._table_name = name self._schema = self._init_schema(schema) self._query_cls = query_cls or Query self._for = None self._for_portion = None if not issubclass(self._query_cls, Query): raise TypeError("Expected 'query_cls' to be subclass of Query") def get_table_name(self) -> str: return self.alias or self._table_name def get_sql(self, **kwargs: Any) -> str: quote_char = kwargs.get("quote_char") table_sql = format_quotes(self._table_name, quote_char) if self._schema is not None: table_sql = "{schema}.{table}".format(schema=self._schema.get_sql(**kwargs), table=table_sql) if self._for: table_sql = "{table} FOR {criterion}".format(table=table_sql, criterion=self._for.get_sql(**kwargs)) elif self._for_portion: table_sql = "{table} FOR PORTION OF {criterion}".format( table=table_sql, criterion=self._for_portion.get_sql(**kwargs) ) return format_alias_sql(table_sql, self.alias, **kwargs) @builder def for_(self, temporal_criterion: Criterion) -> None: if self._for: raise AttributeError("'Query' object already has attribute for_") if self._for_portion: raise AttributeError("'Query' object already has attribute for_portion") self._for = temporal_criterion @builder def for_portion(self, period_criterion: PeriodCriterion) -> None: if self._for_portion: raise AttributeError("'Query' object already has attribute for_portion") if self._for: raise AttributeError("'Query' object already has attribute for_") self._for_portion = period_criterion def __str__(self) -> str: return self.get_sql(quote_char='"') def __eq__(self, other) -> bool: if not isinstance(other, Table): return False if self._table_name != other._table_name: return False if self._schema != other._schema: return False if self.alias != other.alias: return False return True def __repr__(self) -> str: if self._schema: return "Table('{}', schema='{}')".format(self._table_name, self._schema) return "Table('{}')".format(self._table_name) def __ne__(self, other: Any) -> bool: return not self.__eq__(other) def __hash__(self) -> int: return hash(str(self)) def select(self, *terms: int | float | str | bool | Term | Field) -> QueryBuilder: """ Perform a SELECT operation on the current table :param terms: Type: list[expression] A list of terms to select. These can be any type of int, float, str, bool or Term or a Field. :return: QueryBuilder """ return self._query_cls.from_(self).select(*terms) def update(self) -> QueryBuilder: """ Perform an UPDATE operation on the current table :return: QueryBuilder """ return self._query_cls.update(self) def insert(self, *terms: int | float | str | bool | Term | Field) -> QueryBuilder: """ Perform an INSERT operation on the current table :param terms: Type: list[expression] A list of terms to select. These can be any type of int, float, str, bool or any other valid data :return: QueryBuilder """ return self._query_cls.into(self).insert(*terms) def make_tables(*names: tuple[str, str] | str, **kwargs: Any) -> list[Table]: """ Shortcut to create many tables. If `names` param is a tuple, the first position will refer to the `_table_name` while the second will be its `alias`. Any other data structure will be treated as a whole as the `_table_name`. """ tables = [] for name in names: if isinstance(name, tuple) and len(name) == 2: t = Table( name=name[0], alias=name[1], schema=kwargs.get("schema"), query_cls=kwargs.get("query_cls"), ) else: t = Table( name=name, schema=kwargs.get("schema"), query_cls=kwargs.get("query_cls"), ) tables.append(t) return tables class Column: """Represents a column.""" def __init__( self, column_name: str, column_type: str | None = None, nullable: bool | None = None, default: Any | Term | None = None, ) -> None: self.name = column_name self.type = column_type self.nullable = nullable self.default = default if default is None or isinstance(default, Term) else ValueWrapper(default) def get_name_sql(self, **kwargs: Any) -> str: quote_char = kwargs.get("quote_char") column_sql = "{name}".format( name=format_quotes(self.name, quote_char), ) return column_sql def get_sql(self, **kwargs: Any) -> str: column_sql = "{name}{type}{nullable}{default}".format( name=self.get_name_sql(**kwargs), type=" {}".format(self.type) if self.type else "", nullable=" {}".format("NULL" if self.nullable else "NOT NULL") if self.nullable is not None else "", default=" {}".format("DEFAULT " + self.default.get_sql(**kwargs)) if self.default else "", ) return column_sql def __str__(self) -> str: return self.get_sql(quote_char='"') def make_columns(*names: tuple[str, str] | str) -> list[Column]: """ Shortcut to create many columns. If `names` param is a tuple, the first position will refer to the `name` while the second will be its `type`. Any other data structure will be treated as a whole as the `name`. """ columns = [] for name in names: if isinstance(name, tuple) and len(name) == 2: column = Column(column_name=name[0], column_type=name[1]) else: column = Column(column_name=name) columns.append(column) return columns class PeriodFor: def __init__(self, name: str, start_column: str | Column, end_column: str | Column) -> None: self.name = name self.start_column = start_column if isinstance(start_column, Column) else Column(start_column) self.end_column = end_column if isinstance(end_column, Column) else Column(end_column) def get_sql(self, **kwargs: Any) -> str: quote_char = kwargs.get("quote_char") period_for_sql = "PERIOD FOR {name} ({start_column_name},{end_column_name})".format( name=format_quotes(self.name, quote_char), start_column_name=self.start_column.get_name_sql(**kwargs), end_column_name=self.end_column.get_name_sql(**kwargs), ) return period_for_sql # for typing in Query's methods _TableClass = Table class Query: """ Query is the primary class and entry point in pypika. It is used to build queries iteratively using the builder design pattern. This class is immutable. Examples -------- Simple query .. code-block:: python from pypika import Query, Field q = Query.from_('customers').select('*').where(Field("id") == 1) """ @classmethod def _builder(cls, **kwargs: Any) -> QueryBuilder: return QueryBuilder(**kwargs) @classmethod def from_(cls, table: Selectable | str, **kwargs: Any) -> QueryBuilder: """ Query builder entry point. Initializes query building and sets the table to select from. When using this function, the query becomes a SELECT query. :param table: Type: Table or str An instance of a Table object or a string table name. :return: QueryBuilder """ return cls._builder(**kwargs).from_(table) @classmethod def create_table(cls, table: str | Table) -> CreateQueryBuilder: """ Query builder entry point. Initializes query building and sets the table name to be created. When using this function, the query becomes a CREATE statement. :param table: An instance of a Table object or a string table name. :return: CreateQueryBuilder """ return CreateQueryBuilder().create_table(table) @classmethod def create_index(cls, index: str | Index) -> CreateIndexBuilder: """ Query builder entry point. Initializes query building and sets the index name to be created. When using this function, the query becomes a CREATE statement. """ return CreateIndexBuilder().create_index(index) @classmethod def drop_database(cls, database: Database | Table) -> DropQueryBuilder: """ Query builder entry point. Initializes query building and sets the table name to be dropped. When using this function, the query becomes a DROP statement. :param database: An instance of a Database object or a string database name. :return: DropQueryBuilder """ return DropQueryBuilder().drop_database(database) @classmethod def drop_table(cls, table: str | Table) -> DropQueryBuilder: """ Query builder entry point. Initializes query building and sets the table name to be dropped. When using this function, the query becomes a DROP statement. :param table: An instance of a Table object or a string table name. :return: DropQueryBuilder """ return DropQueryBuilder().drop_table(table) @classmethod def drop_user(cls, user: str) -> DropQueryBuilder: """ Query builder entry point. Initializes query building and sets the table name to be dropped. When using this function, the query becomes a DROP statement. :param user: String user name. :return: DropQueryBuilder """ return DropQueryBuilder().drop_user(user) @classmethod def drop_view(cls, view: str) -> DropQueryBuilder: """ Query builder entry point. Initializes query building and sets the table name to be dropped. When using this function, the query becomes a DROP statement. :param view: String view name. :return: DropQueryBuilder """ return DropQueryBuilder().drop_view(view) @classmethod def drop_index(cls, index: str | Index) -> DropQueryBuilder: """ Query builder entry point. Initializes query building and sets the index name to be dropped. When using this function, the query becomes a DROP statement. """ return DropQueryBuilder().drop_index(index) @classmethod def into(cls, table: Table | str, **kwargs: Any) -> QueryBuilder: """ Query builder entry point. Initializes query building and sets the table to insert into. When using this function, the query becomes an INSERT query. :param table: Type: Table or str An instance of a Table object or a string table name. :return QueryBuilder """ return cls._builder(**kwargs).into(table) @classmethod def with_(cls, table: str | Selectable, name: str, **kwargs: Any) -> QueryBuilder: return cls._builder(**kwargs).with_(table, name) @classmethod def select(cls, *terms: int | float | str | bool | Term, **kwargs: Any) -> QueryBuilder: """ Query builder entry point. Initializes query building without a table and selects fields. Useful when testing SQL functions. :param terms: Type: list[expression] A list of terms to select. These can be any type of int, float, str, bool, or Term. They cannot be a Field unless the function ``Query.from_`` is called first. :return: QueryBuilder """ return cls._builder(**kwargs).select(*terms) @classmethod def update(cls, table: str | Table, **kwargs) -> QueryBuilder: """ Query builder entry point. Initializes query building and sets the table to update. When using this function, the query becomes an UPDATE query. :param table: Type: Table or str An instance of a Table object or a string table name. :return: QueryBuilder """ return cls._builder(**kwargs).update(table) @classmethod def Table(cls, table_name: str, **kwargs) -> _TableClass: """ Convenience method for creating a Table that uses this Query class. :param table_name: Type: str A string table name. :return: Table """ kwargs["query_cls"] = cls return Table(table_name, **kwargs) @classmethod def Tables(cls, *names: tuple[str, str] | str, **kwargs: Any) -> list[_TableClass]: """ Convenience method for creating many tables that uses this Query class. See ``Query.make_tables`` for details. :param names: Type: list[str or tuple] A list of string table names, or name and alias tuples. :return: Table """ kwargs["query_cls"] = cls return make_tables(*names, **kwargs) class _SetOperation(Selectable, Term): """ A Query class wrapper for a all set operations, Union DISTINCT or ALL, Intersect, Except or Minus Created via the functions `Query.union`,`Query.union_all`,`Query.intersect`, `Query.except_of`,`Query.minus`. This class should not be instantiated directly. """ def __init__( self, base_query: QueryBuilder, set_operation_query: QueryBuilder, set_operation: SetOperation, alias: str | None = None, wrapper_cls: type[ValueWrapper] = ValueWrapper, ): super().__init__(alias) self.base_query = base_query self._set_operation = [(set_operation, set_operation_query)] self._orderbys = [] self._limit = None self._offset = None self._wrapper_cls = wrapper_cls @builder def orderby(self, *fields: Field, **kwargs: Any) -> None: for field in fields: field = ( Field(field, table=self.base_query._from[0]) if isinstance(field, str) else self.base_query.wrap_constant(field) ) self._orderbys.append((field, kwargs.get("order"))) @builder def limit(self, limit: int) -> None: self._limit = limit @builder def offset(self, offset: int) -> None: self._offset = offset @builder def union(self, other: Selectable) -> None: self._set_operation.append((SetOperation.union, other)) @builder def union_all(self, other: Selectable) -> None: self._set_operation.append((SetOperation.union_all, other)) @builder def intersect(self, other: Selectable) -> None: self._set_operation.append((SetOperation.intersect, other)) @builder def except_of(self, other: Selectable) -> None: self._set_operation.append((SetOperation.except_of, other)) @builder def minus(self, other: Selectable) -> None: self._set_operation.append((SetOperation.minus, other)) def __add__(self, other: Selectable) -> _SetOperation: return self.union(other) def __mul__(self, other: Selectable) -> _SetOperation: return self.union_all(other) def __sub__(self, other: QueryBuilder) -> _SetOperation: return self.minus(other) def __str__(self) -> str: return self.get_sql() def get_sql(self, with_alias: bool = False, subquery: bool = False, **kwargs: Any) -> str: set_operation_template = " {type} {query_string}" kwargs.setdefault("dialect", self.base_query.dialect) # This initializes the quote char based on the base query, which could be a dialect specific query class # This might be overridden if quote_char is set explicitly in kwargs kwargs.setdefault("quote_char", self.base_query.QUOTE_CHAR) base_querystring = self.base_query.get_sql(subquery=self.base_query.wrap_set_operation_queries, **kwargs) querystring = base_querystring for set_operation, set_operation_query in self._set_operation: set_operation_querystring = set_operation_query.get_sql( subquery=self.base_query.wrap_set_operation_queries, **kwargs ) if len(self.base_query._selects) != len(set_operation_query._selects): raise SetOperationException( "Queries must have an equal number of select statements in a set operation." "\n\nMain Query:\n{query1}\n\nSet Operations Query:\n{query2}".format( query1=base_querystring, query2=set_operation_querystring ) ) querystring += set_operation_template.format( type=set_operation.value, query_string=set_operation_querystring ) if self._orderbys: querystring += self._orderby_sql(**kwargs) if self._limit is not None: querystring += self._limit_sql() if self._offset: querystring += self._offset_sql() if subquery: querystring = "({query})".format(query=querystring, **kwargs) if with_alias: return format_alias_sql(querystring, self.alias or self._table_name, **kwargs) return querystring def _orderby_sql(self, quote_char: str | None = None, **kwargs: Any) -> str: """ Produces the ORDER BY part of the query. This is a list of fields and possibly their directionality, ASC or DESC. The clauses are stored in the query under self._orderbys as a list of tuples containing the field and directionality (which can be None). If an order by field is used in the select clause, determined by a matching , then the ORDER BY clause will use the alias, otherwise the field will be rendered as SQL. """ clauses = [] selected_aliases = {s.alias for s in self.base_query._selects} for field, directionality in self._orderbys: term = ( format_quotes(field.alias, quote_char) if field.alias and field.alias in selected_aliases else field.get_sql(quote_char=quote_char, **kwargs) ) clauses.append( "{term} {orient}".format(term=term, orient=directionality.value) if directionality is not None else term ) return " ORDER BY {orderby}".format(orderby=",".join(clauses)) def _offset_sql(self) -> str: return " OFFSET {offset}".format(offset=self._offset) def _limit_sql(self) -> str: return " LIMIT {limit}".format(limit=self._limit) class QueryBuilder(Selectable, Term): """ Query Builder is the main class in pypika which stores the state of a query and offers functions which allow the state to be branched immutably. """ QUOTE_CHAR = '"' SECONDARY_QUOTE_CHAR = "'" ALIAS_QUOTE_CHAR = None QUERY_ALIAS_QUOTE_CHAR = None QUERY_CLS = Query def __init__( self, dialect: Dialects | None = None, wrap_set_operation_queries: bool = True, wrapper_cls: type[ValueWrapper] = ValueWrapper, immutable: bool = True, as_keyword: bool = False, ): super().__init__(None) self._from = [] self._insert_table = None self._update_table = None self._delete_from = False self._replace = False self._with = [] self._selects = [] self._force_indexes = [] self._use_indexes = [] self._columns = [] self._values = [] self._distinct = False self._ignore = False self._for_update = False self._wheres = None self._prewheres = None self._groupbys = [] self._with_totals = False self._havings = None self._qualifys = None self._orderbys = [] self._joins = [] self._unions = [] self._using = [] self._limit = None self._offset = None self._updates = [] self._select_star = False self._select_star_tables = set() self._mysql_rollup = False self._select_into = False self._subquery_count = 0 self._foreign_table = False self.dialect = dialect self.as_keyword = as_keyword self.wrap_set_operation_queries = wrap_set_operation_queries self._wrapper_cls = wrapper_cls self.immutable = immutable def __copy__(self) -> QueryBuilder: newone = type(self).__new__(type(self)) newone.__dict__.update(self.__dict__) newone._select_star_tables = copy(self._select_star_tables) newone._from = copy(self._from) newone._with = copy(self._with) newone._selects = copy(self._selects) newone._columns = copy(self._columns) newone._values = copy(self._values) newone._groupbys = copy(self._groupbys) newone._orderbys = copy(self._orderbys) newone._joins = copy(self._joins) newone._unions = copy(self._unions) newone._updates = copy(self._updates) newone._force_indexes = copy(self._force_indexes) newone._use_indexes = copy(self._use_indexes) return newone @builder def from_(self, selectable: Selectable | Query | str) -> None: """ Adds a table to the query. This function can only be called once and will raise an AttributeError if called a second time. :param selectable: Type: ``Table``, ``Query``, or ``str`` When a ``str`` is passed, a table with the name matching the ``str`` value is used. :returns A copy of the query with the table added. """ self._from.append(Table(selectable) if isinstance(selectable, str) else selectable) if isinstance(selectable, (QueryBuilder, _SetOperation)) and selectable.alias is None: if isinstance(selectable, QueryBuilder): sub_query_count = selectable._subquery_count else: sub_query_count = 0 sub_query_count = max(self._subquery_count, sub_query_count) selectable.alias = "sq%d" % sub_query_count self._subquery_count = sub_query_count + 1 @builder def replace_table(self, current_table: Table | None, new_table: Table | None) -> None: """ Replaces all occurrences of the specified table with the new table. Useful when reusing fields across queries. :param current_table: The table instance to be replaces. :param new_table: The table instance to replace with. :return: A copy of the query with the tables replaced. """ self._from = [new_table if table == current_table else table for table in self._from] self._insert_table = new_table if self._insert_table == current_table else self._insert_table self._update_table = new_table if self._update_table == current_table else self._update_table self._with = [alias_query.replace_table(current_table, new_table) for alias_query in self._with] self._selects = [select.replace_table(current_table, new_table) for select in self._selects] self._columns = [column.replace_table(current_table, new_table) for column in self._columns] self._values = [ [value.replace_table(current_table, new_table) for value in value_list] for value_list in self._values ] self._wheres = self._wheres.replace_table(current_table, new_table) if self._wheres else None self._prewheres = self._prewheres.replace_table(current_table, new_table) if self._prewheres else None self._groupbys = [groupby.replace_table(current_table, new_table) for groupby in self._groupbys] self._havings = self._havings.replace_table(current_table, new_table) if self._havings else None self._qualifys = self._qualifys.replace_table(current_table, new_table) if self._qualifys else None self._orderbys = [ (orderby[0].replace_table(current_table, new_table), orderby[1]) for orderby in self._orderbys ] self._joins = [join.replace_table(current_table, new_table) for join in self._joins] if current_table in self._select_star_tables: self._select_star_tables.remove(current_table) self._select_star_tables.add(new_table) @builder def with_(self, selectable: Selectable, name: str) -> None: t = AliasedQuery(name, selectable) self._with.append(t) @builder def into(self, table: str | Table) -> None: if self._insert_table is not None: raise AttributeError("'Query' object has no attribute '%s'" % "into") if self._selects: self._select_into = True self._insert_table = table if isinstance(table, Table) else Table(table) @builder def select(self, *terms: Any) -> None: for term in terms: if isinstance(term, Field): self._select_field(term) elif isinstance(term, str): self._select_field_str(term) elif isinstance(term, (Function, ArithmeticExpression)): self._select_other(term) else: self._select_other(self.wrap_constant(term, wrapper_cls=self._wrapper_cls)) @builder def delete(self) -> None: if self._delete_from or self._selects or self._update_table: raise AttributeError("'Query' object has no attribute '%s'" % "delete") self._delete_from = True @builder def update(self, table: str | Table) -> None: if self._update_table is not None or self._selects or self._delete_from: raise AttributeError("'Query' object has no attribute '%s'" % "update") self._update_table = table if isinstance(table, Table) else Table(table) @builder def columns(self, *terms: Any) -> None: if self._insert_table is None: raise AttributeError("'Query' object has no attribute '%s'" % "insert") if terms and isinstance(terms[0], (list, tuple)): terms = terms[0] for term in terms: if isinstance(term, str): term = Field(term, table=self._insert_table) self._columns.append(term) @builder def insert(self, *terms: Any) -> None: self._apply_terms(*terms) self._replace = False @builder def replace(self, *terms: Any) -> None: self._apply_terms(*terms) self._replace = True @builder def force_index(self, term: str | Index, *terms: str | Index) -> None: for t in (term, *terms): if isinstance(t, Index): self._force_indexes.append(t) elif isinstance(t, str): self._force_indexes.append(Index(t)) @builder def use_index(self, term: str | Index, *terms: str | Index) -> None: for t in (term, *terms): if isinstance(t, Index): self._use_indexes.append(t) elif isinstance(t, str): self._use_indexes.append(Index(t)) @builder def distinct(self) -> None: self._distinct = True @builder def for_update(self) -> None: self._for_update = True @builder def ignore(self) -> None: self._ignore = True @builder def prewhere(self, criterion: Criterion) -> None: if not self._validate_table(criterion): self._foreign_table = True if self._prewheres: self._prewheres &= criterion else: self._prewheres = criterion @builder def where(self, criterion: Term | EmptyCriterion) -> None: if isinstance(criterion, EmptyCriterion): return if not self._validate_table(criterion): self._foreign_table = True if self._wheres: self._wheres &= criterion else: self._wheres = criterion @builder def having(self, criterion: Term | EmptyCriterion) -> None: if isinstance(criterion, EmptyCriterion): return if self._havings: self._havings &= criterion else: self._havings = criterion @builder def qualify(self, criterion: Term | EmptyCriterion) -> None: if isinstance(criterion, EmptyCriterion): return if self._qualifys: self._qualifys &= criterion else: self._qualifys = criterion @builder def groupby(self, *terms: str | int | Term) -> None: for term in terms: if isinstance(term, str): term = Field(term, table=self._from[0]) elif isinstance(term, int): term = Field(str(term), table=self._from[0]).wrap_constant(term) self._groupbys.append(term) @builder def with_totals(self) -> None: self._with_totals = True @builder def rollup(self, *terms: list | tuple | set | Term, **kwargs: Any) -> None: for_mysql = "mysql" == kwargs.get("vendor") if self._mysql_rollup: raise AttributeError("'Query' object has no attribute '%s'" % "rollup") terms = [Tuple(*term) if isinstance(term, (list, tuple, set)) else term for term in terms] if for_mysql: # MySQL rolls up all of the dimensions always if not terms and not self._groupbys: raise RollupException( "At least one group is required. Call Query.groupby(term) or pass" "as parameter to rollup." ) self._mysql_rollup = True self._groupbys += terms elif 0 < len(self._groupbys) and isinstance(self._groupbys[-1], Rollup): # If a rollup was added last, then append the new terms to the previous rollup self._groupbys[-1].args += terms else: self._groupbys.append(Rollup(*terms)) @builder def orderby(self, *fields: Any, **kwargs: Any) -> None: for field in fields: field = Field(field, table=self._from[0]) if isinstance(field, str) else self.wrap_constant(field) self._orderbys.append((field, kwargs.get("order"))) @builder def join( self, item: Table | QueryBuilder | AliasedQuery | Selectable, how: JoinType = JoinType.inner ) -> Joiner[Self]: if isinstance(item, Table): return Joiner(self, item, how, type_label="table") elif isinstance(item, QueryBuilder): if item.alias is None: self._tag_subquery(item) return Joiner(self, item, how, type_label="subquery") elif isinstance(item, AliasedQuery): return Joiner(self, item, how, type_label="table") elif isinstance(item, Selectable): return Joiner(self, item, how, type_label="subquery") raise ValueError("Cannot join on type '%s'" % type(item)) def inner_join(self, item: Table | QueryBuilder | AliasedQuery) -> Joiner[Self]: return self.join(item, JoinType.inner) def left_join(self, item: Table | QueryBuilder | AliasedQuery) -> Joiner[Self]: return self.join(item, JoinType.left) def left_outer_join(self, item: Table | QueryBuilder | AliasedQuery) -> Joiner[Self]: return self.join(item, JoinType.left_outer) def right_join(self, item: Table | QueryBuilder | AliasedQuery) -> Joiner[Self]: return self.join(item, JoinType.right) def right_outer_join(self, item: Table | QueryBuilder | AliasedQuery) -> Joiner[Self]: return self.join(item, JoinType.right_outer) def outer_join(self, item: Table | QueryBuilder | AliasedQuery) -> Joiner[Self]: return self.join(item, JoinType.outer) def full_outer_join(self, item: Table | QueryBuilder | AliasedQuery) -> Joiner[Self]: return self.join(item, JoinType.full_outer) def cross_join(self, item: Table | QueryBuilder | AliasedQuery) -> Joiner[Self]: return self.join(item, JoinType.cross) def hash_join(self, item: Table | QueryBuilder | AliasedQuery) -> Joiner[Self]: return self.join(item, JoinType.hash) @builder def limit(self, limit: int) -> None: self._limit = limit @builder def offset(self, offset: int) -> None: self._offset = offset @builder def union(self, other: QueryBuilder) -> _SetOperation: return _SetOperation(self, other, SetOperation.union, wrapper_cls=self._wrapper_cls) @builder def union_all(self, other: QueryBuilder) -> _SetOperation: return _SetOperation(self, other, SetOperation.union_all, wrapper_cls=self._wrapper_cls) @builder def intersect(self, other: QueryBuilder) -> _SetOperation: return _SetOperation(self, other, SetOperation.intersect, wrapper_cls=self._wrapper_cls) @builder def except_of(self, other: QueryBuilder) -> _SetOperation: return _SetOperation(self, other, SetOperation.except_of, wrapper_cls=self._wrapper_cls) @builder def minus(self, other: QueryBuilder) -> _SetOperation: return _SetOperation(self, other, SetOperation.minus, wrapper_cls=self._wrapper_cls) @builder def set(self, field: Field | str, value: Any) -> None: field = Field(field) if not isinstance(field, Field) else field if not isinstance(value, Term): value = self.wrap_constant(value, wrapper_cls=self._wrapper_cls) self._updates.append((field, value)) def __add__(self, other: QueryBuilder) -> _SetOperation: return self.union(other) def __mul__(self, other: QueryBuilder) -> _SetOperation: return self.union_all(other) def __sub__(self, other: QueryBuilder) -> _SetOperation: return self.minus(other) @builder def slice(self, slice: slice) -> None: self._offset = slice.start self._limit = slice.stop def __getitem__(self, item: Any) -> QueryBuilder | Field: if not isinstance(item, slice): return super().__getitem__(item) return self.slice(item) @staticmethod def _list_aliases(field_set: Sequence[Field], quote_char: str | None = None) -> list[str]: return [field.alias or field.get_sql(quote_char=quote_char) for field in field_set] def _select_field_str(self, term: str) -> None: if 0 == len(self._from): raise QueryException(f"Cannot select {term}, no FROM table specified.") if term == "*": self._select_star = True self._selects = [Star()] return self._select_field(Field(term, table=self._from[0])) def _select_field(self, term: Field) -> None: if self._select_star: # Do not add select terms after a star is selected return if term.table in self._select_star_tables: # Do not add select terms for table after a table star is selected return if isinstance(term, Star): self._selects = [ select for select in self._selects if not hasattr(select, "table") or term.table != select.table ] self._select_star_tables.add(term.table) self._selects.append(term) def _select_other(self, function: Function) -> None: self._selects.append(function) def fields_(self) -> list[Field]: # Don't return anything here. Subqueries have their own fields. return [] def do_join(self, join: Join) -> None: base_tables = self._from + [self._update_table] + self._with join.validate(base_tables, self._joins) table_in_query = any(isinstance(clause, Table) and join.item in base_tables for clause in base_tables) if isinstance(join.item, Table) and join.item.alias is None and table_in_query: # On the odd chance that we join the same table as the FROM table and don't set an alias # FIXME only works once join.item.alias = join.item._table_name + "2" self._joins.append(join) def is_joined(self, table: Table) -> bool: return any(table == join.item for join in self._joins) def _validate_table(self, term: Term) -> bool: """ Returns False if the term references a table not already part of the FROM clause or JOINS and True otherwise. """ base_tables = self._from + [self._update_table] for field in term.fields_(): table_in_base_tables = field.table in base_tables table_in_joins = field.table in [join.item for join in self._joins] if all( [ field.table is not None, not table_in_base_tables, not table_in_joins, field.table != self._update_table, ] ): return False return True def _tag_subquery(self, subquery: QueryBuilder) -> None: subquery.alias = "sq%d" % self._subquery_count self._subquery_count += 1 def _apply_terms(self, *terms: Any) -> None: """ Handy function for INSERT and REPLACE statements in order to check if terms are introduced and how append them to `self._values` """ if self._insert_table is None: raise AttributeError("'Query' object has no attribute '%s'" % "insert") if not terms: return if not isinstance(terms[0], (list, tuple, set)): terms = [terms] for values in terms: self._values.append([value if isinstance(value, Term) else self.wrap_constant(value) for value in values]) def __str__(self) -> str: return self.get_sql(dialect=self.dialect) def __repr__(self) -> str: return self.__str__() def __eq__(self, other: QueryBuilder) -> bool: if not isinstance(other, QueryBuilder): return False if not self.alias == other.alias: return False return True def __ne__(self, other: QueryBuilder) -> bool: return not self.__eq__(other) def __hash__(self) -> int: return hash(self.alias) + sum(hash(clause) for clause in self._from) def _set_kwargs_defaults(self, kwargs: dict) -> None: kwargs.setdefault("quote_char", self.QUOTE_CHAR) kwargs.setdefault("secondary_quote_char", self.SECONDARY_QUOTE_CHAR) kwargs.setdefault("alias_quote_char", self.ALIAS_QUOTE_CHAR) kwargs.setdefault("as_keyword", self.as_keyword) kwargs.setdefault("dialect", self.dialect) def get_sql(self, with_alias: bool = False, subquery: bool = False, **kwargs: Any) -> str: self._set_kwargs_defaults(kwargs) if not (self._selects or self._insert_table or self._delete_from or self._update_table): return "" if self._insert_table and not (self._selects or self._values): return "" if self._update_table and not self._updates: return "" has_joins = bool(self._joins) has_multiple_from_clauses = 1 < len(self._from) has_subquery_from_clause = 0 < len(self._from) and isinstance(self._from[0], QueryBuilder) has_reference_to_foreign_table = self._foreign_table has_update_from = self._update_table and self._from kwargs["with_namespace"] = any( [ has_joins, has_multiple_from_clauses, has_subquery_from_clause, has_reference_to_foreign_table, has_update_from, ] ) if self._update_table: querystring = self._with_sql(**kwargs) if self._with else "" querystring += self._update_sql(**kwargs) if self._joins: querystring += " " + " ".join(join.get_sql(**kwargs) for join in self._joins) querystring += self._set_sql(**kwargs) if self._from: querystring += self._from_sql(**kwargs) if self._wheres: querystring += self._where_sql(**kwargs) if self._limit is not None: querystring += self._limit_sql() return querystring if self._delete_from: querystring = self._delete_sql(**kwargs) elif not self._select_into and self._insert_table: querystring = self._with_sql(**kwargs) if self._with else "" if self._replace: querystring += self._replace_sql(**kwargs) else: querystring += self._insert_sql(**kwargs) if self._columns: querystring += self._columns_sql(**kwargs) if self._values: querystring += self._values_sql(**kwargs) return querystring else: querystring += " " + self._select_sql(**kwargs) else: querystring = self._with_sql(**kwargs) if self._with else "" querystring += self._select_sql(**kwargs) if self._insert_table: querystring += self._into_sql(**kwargs) if self._from: querystring += self._from_sql(**kwargs) if self._using: querystring += self._using_sql(**kwargs) if self._force_indexes: querystring += self._force_index_sql(**kwargs) if self._use_indexes: querystring += self._use_index_sql(**kwargs) if self._joins: querystring += " " + " ".join(join.get_sql(**kwargs) for join in self._joins) if self._prewheres: querystring += self._prewhere_sql(**kwargs) if self._wheres: querystring += self._where_sql(**kwargs) if self._groupbys: querystring += self._group_sql(**kwargs) if self._mysql_rollup: querystring += self._rollup_sql() if self._havings: querystring += self._having_sql(**kwargs) if self._qualifys: querystring += self._qualify_sql(**kwargs) if self._orderbys: querystring += self._orderby_sql(**kwargs) querystring = self._apply_pagination(querystring, **kwargs) if self._for_update: querystring += self._for_update_sql(**kwargs) if subquery: querystring = "({query})".format(query=querystring) if with_alias: kwargs['alias_quote_char'] = ( self.ALIAS_QUOTE_CHAR if self.QUERY_ALIAS_QUOTE_CHAR is None else self.QUERY_ALIAS_QUOTE_CHAR ) return format_alias_sql(querystring, self.alias, **kwargs) return querystring def _apply_pagination(self, querystring: str, **kwargs) -> str: if self._limit is not None: querystring += self._limit_sql() if self._offset: querystring += self._offset_sql() return querystring def _with_sql(self, **kwargs: Any) -> str: return "WITH " + ",".join( clause.name + " AS (" + clause.get_sql(subquery=False, with_alias=False, **kwargs) + ") " for clause in self._with ) def _distinct_sql(self, **kwargs: Any) -> str: distinct = 'DISTINCT ' if self._distinct else '' return distinct def _for_update_sql(self, **kwargs) -> str: for_update = ' FOR UPDATE' if self._for_update else '' return for_update def _select_sql(self, **kwargs: Any) -> str: return "SELECT {distinct}{select}".format( distinct=self._distinct_sql(**kwargs), select=",".join(term.get_sql(with_alias=True, subquery=True, **kwargs) for term in self._selects), ) def _insert_sql(self, **kwargs: Any) -> str: return "INSERT {ignore}INTO {table}".format( table=self._insert_table.get_sql(**kwargs), ignore="IGNORE " if self._ignore else "", ) def _replace_sql(self, **kwargs: Any) -> str: return "REPLACE INTO {table}".format( table=self._insert_table.get_sql(**kwargs), ) @staticmethod def _delete_sql(**kwargs: Any) -> str: return "DELETE" def _update_sql(self, **kwargs: Any) -> str: return "UPDATE {table}".format(table=self._update_table.get_sql(**kwargs)) def _columns_sql(self, with_namespace: bool = False, **kwargs: Any) -> str: """ SQL for Columns clause for INSERT queries :param with_namespace: Remove from kwargs, never format the column terms with namespaces since only one table can be inserted into """ return " ({columns})".format( columns=",".join(term.get_sql(with_namespace=False, **kwargs) for term in self._columns) ) def _values_sql(self, **kwargs: Any) -> str: return " VALUES ({values})".format( values="),(".join( ",".join(term.get_sql(with_alias=True, subquery=True, **kwargs) for term in row) for row in self._values ) ) def _into_sql(self, **kwargs: Any) -> str: return " INTO {table}".format( table=self._insert_table.get_sql(with_alias=False, **kwargs), ) def _from_sql(self, with_namespace: bool = False, **kwargs: Any) -> str: return " FROM {selectable}".format( selectable=",".join(clause.get_sql(subquery=True, with_alias=True, **kwargs) for clause in self._from) ) def _using_sql(self, with_namespace: bool = False, **kwargs: Any) -> str: return " USING {selectable}".format( selectable=",".join(clause.get_sql(subquery=True, with_alias=True, **kwargs) for clause in self._using) ) def _force_index_sql(self, **kwargs: Any) -> str: return " FORCE INDEX ({indexes})".format( indexes=",".join(index.get_sql(**kwargs) for index in self._force_indexes), ) def _use_index_sql(self, **kwargs: Any) -> str: return " USE INDEX ({indexes})".format( indexes=",".join(index.get_sql(**kwargs) for index in self._use_indexes), ) def _prewhere_sql(self, quote_char: str | None = None, **kwargs: Any) -> str: return " PREWHERE {prewhere}".format( prewhere=self._prewheres.get_sql(quote_char=quote_char, subquery=True, **kwargs) ) def _where_sql(self, quote_char: str | None = None, **kwargs: Any) -> str: return " WHERE {where}".format(where=self._wheres.get_sql(quote_char=quote_char, subquery=True, **kwargs)) def _group_sql( self, quote_char: str | None = None, alias_quote_char: str | None = None, groupby_alias: bool = True, **kwargs: Any, ) -> str: """ Produces the GROUP BY part of the query. This is a list of fields. The clauses are stored in the query under self._groupbys as a list fields. If an groupby field is used in the select clause, determined by a matching alias, and the groupby_alias is set True then the GROUP BY clause will use the alias, otherwise the entire field will be rendered as SQL. """ clauses = [] selected_aliases = {s.alias for s in self._selects} for field in self._groupbys: if groupby_alias and field.alias and field.alias in selected_aliases: clauses.append(format_quotes(field.alias, alias_quote_char or quote_char)) else: clauses.append(field.get_sql(quote_char=quote_char, alias_quote_char=alias_quote_char, **kwargs)) sql = " GROUP BY {groupby}".format(groupby=",".join(clauses)) if self._with_totals: return sql + " WITH TOTALS" return sql def _orderby_sql( self, quote_char: str | None = None, alias_quote_char: str | None = None, orderby_alias: bool = True, **kwargs: Any, ) -> str: """ Produces the ORDER BY part of the query. This is a list of fields and possibly their directionality, ASC or DESC. The clauses are stored in the query under self._orderbys as a list of tuples containing the field and directionality (which can be None). If an order by field is used in the select clause, determined by a matching, and the orderby_alias is set True then the ORDER BY clause will use the alias, otherwise the field will be rendered as SQL. """ clauses = [] selected_aliases = {s.alias for s in self._selects} for field, directionality in self._orderbys: term = ( format_quotes(field.alias, alias_quote_char or quote_char) if orderby_alias and field.alias and field.alias in selected_aliases else field.get_sql(quote_char=quote_char, alias_quote_char=alias_quote_char, **kwargs) ) clauses.append( "{term} {orient}".format(term=term, orient=directionality.value) if directionality is not None else term ) return " ORDER BY {orderby}".format(orderby=",".join(clauses)) def _rollup_sql(self) -> str: return " WITH ROLLUP" def _having_sql(self, quote_char: str | None = None, **kwargs: Any) -> str: return " HAVING {having}".format(having=self._havings.get_sql(quote_char=quote_char, **kwargs)) def _qualify_sql(self, quote_char: str | None = None, **kwargs: Any) -> str: return " QUALIFY {qualify}".format(qualify=self._qualifys.get_sql(quote_char=quote_char, **kwargs)) def _offset_sql(self) -> str: return f" OFFSET {self._offset}" def _limit_sql(self) -> str: return f" LIMIT {self._limit}" def _set_sql(self, **kwargs: Any) -> str: return " SET {set}".format( set=",".join( "{field}={value}".format( field=field.get_sql(**dict(kwargs, with_namespace=False)), value=value.get_sql(**kwargs) ) for field, value in self._updates ) ) def pipe(self, func, *args, **kwargs): """Call a function on the current object and return the result. Example usage: .. code-block:: python from pypika import Query, functions as fn from pypika.queries import QueryBuilder def rows_by_group(query: QueryBuilder, *groups) -> QueryBuilder: return ( query .select(*groups, fn.Count("*").as_("n_rows")) .groupby(*groups) ) base_query = Query.from_("table") col1_agg = base_query.pipe(rows_by_group, "col1") col2_agg = base_query.pipe(rows_by_group, "col2") col1_col2_agg = base_query.pipe(rows_by_group, "col1", "col2") Makes chaining functions together easier, especially when the functions are defined elsewhere. For example, you could define a function that filters rows by a date range and then group by a set of columns: .. code-block:: python from datetime import datetime, timedelta from pypika import Field def days_since(query: QueryBuilder, n_days: int) -> QueryBuilder: return ( query .where("date" > fn.Date(datetime.now().date() - timedelta(days=n_days))) ) ( base_query .pipe(days_since, n_days=7) .pipe(rows_by_group, "col1", "col2") ) """ return func(self, *args, **kwargs) class Joiner(Generic[QB]): def __init__(self, query: QB, item: Table | QueryBuilder | AliasedQuery, how: JoinType, type_label: str) -> None: self.query = query self.item = item self.how = how self.type_label = type_label def on(self, criterion: Criterion | None, collate: str | None = None) -> QB: if criterion is None: raise JoinException( "Parameter 'criterion' is required for a " "{type} JOIN but was not supplied.".format(type=self.type_label) ) self.query.do_join(JoinOn(self.item, self.how, criterion, collate)) return self.query def on_field(self, *fields: Any) -> QB: if not fields: raise JoinException( "Parameter 'fields' is required for a " "{type} JOIN but was not supplied.".format(type=self.type_label) ) criterion = None for field in fields: constituent = Field(field, table=self.query._from[0]) == Field(field, table=self.item) criterion = constituent if criterion is None else criterion & constituent self.query.do_join(JoinOn(self.item, self.how, criterion)) return self.query def using(self, *fields: Any) -> QB: if not fields: raise JoinException("Parameter 'fields' is required when joining with a using clause but was not supplied.") self.query.do_join(JoinUsing(self.item, self.how, [Field(field) for field in fields])) return self.query def cross(self) -> QB: """Return cross join""" self.query.do_join(Join(self.item, JoinType.cross)) return self.query class Join: def __init__(self, item: Term, how: JoinType) -> None: self.item = item self.how = how def get_sql(self, **kwargs: Any) -> str: sql = "JOIN {table}".format( table=self.item.get_sql(subquery=True, with_alias=True, **kwargs), ) if self.how.value: return "{type} {join}".format(join=sql, type=self.how.value) return sql def validate(self, _from: Sequence[Table], _joins: Sequence[Table]) -> None: pass @builder def replace_table(self, current_table: Table | None, new_table: Table | None) -> None: """ Replaces all occurrences of the specified table with the new table. Useful when reusing fields across queries. :param current_table: The table to be replaced. :param new_table: The table to replace with. :return: A copy of the join with the tables replaced. """ self.item = self.item.replace_table(current_table, new_table) class JoinOn(Join): def __init__(self, item: Term, how: JoinType, criteria: QueryBuilder, collate: str | None = None) -> None: super().__init__(item, how) self.criterion = criteria self.collate = collate def get_sql(self, **kwargs: Any) -> str: join_sql = super().get_sql(**kwargs) return "{join} ON {criterion}{collate}".format( join=join_sql, criterion=self.criterion.get_sql(subquery=True, **kwargs), collate=f" COLLATE {self.collate}" if self.collate else "", ) def validate(self, _from: Sequence[Table], _joins: Sequence[Table]) -> None: criterion_tables = set([f.table for f in self.criterion.fields_()]) available_tables = set(_from) | {join.item for join in _joins} | {self.item} missing_tables = criterion_tables - available_tables if missing_tables: raise JoinException( "Invalid join criterion. One field is required from the joined item and " "another from the selected table or an existing join. Found [{tables}]".format( tables=", ".join(map(str, missing_tables)) ) ) @builder def replace_table(self, current_table: Table | None, new_table: Table | None) -> None: """ Replaces all occurrences of the specified table with the new table. Useful when reusing fields across queries. :param current_table: The table to be replaced. :param new_table: The table to replace with. :return: A copy of the join with the tables replaced. """ self.item = new_table if self.item == current_table else self.item self.criterion = self.criterion.replace_table(current_table, new_table) class JoinUsing(Join): def __init__(self, item: Term, how: JoinType, fields: Sequence[Field]) -> None: super().__init__(item, how) self.fields = fields def get_sql(self, **kwargs: Any) -> str: join_sql = super().get_sql(**kwargs) return "{join} USING ({fields})".format( join=join_sql, fields=",".join(field.get_sql(**kwargs) for field in self.fields), ) def validate(self, _from: Sequence[Table], _joins: Sequence[Table]) -> None: pass @builder def replace_table(self, current_table: Table | None, new_table: Table | None) -> None: """ Replaces all occurrences of the specified table with the new table. Useful when reusing fields across queries. :param current_table: The table to be replaced. :param new_table: The table to replace with. :return: A copy of the join with the tables replaced. """ self.item = new_table if self.item == current_table else self.item self.fields = [field.replace_table(current_table, new_table) for field in self.fields] class CreateQueryBuilder: """ Query builder used to build CREATE queries. """ QUOTE_CHAR = '"' SECONDARY_QUOTE_CHAR = "'" ALIAS_QUOTE_CHAR = None QUERY_CLS = Query def __init__(self, dialect: Dialects | None = None) -> None: self._create_table = None self._temporary = False self._unlogged = False self._as_select = None self._columns = [] self._period_fors = [] self._with_system_versioning = False self._primary_key = None self._uniques = [] self._if_not_exists = False self.dialect = dialect self._foreign_key = None self._foreign_key_reference_table = None self._foreign_key_reference = None self._foreign_key_on_update: ReferenceOption = None self._foreign_key_on_delete: ReferenceOption = None def _set_kwargs_defaults(self, kwargs: dict) -> None: kwargs.setdefault("quote_char", self.QUOTE_CHAR) kwargs.setdefault("secondary_quote_char", self.SECONDARY_QUOTE_CHAR) kwargs.setdefault("dialect", self.dialect) @builder def create_table(self, table: Table | str) -> None: """ Creates the table. :param table: An instance of a Table object or a string table name. :raises AttributeError: If the table is already created. :return: CreateQueryBuilder. """ if self._create_table: raise AttributeError("'Query' object already has attribute create_table") self._create_table = table if isinstance(table, Table) else Table(table) @builder def temporary(self) -> None: """ Makes the table temporary. :return: CreateQueryBuilder. """ self._temporary = True @builder def unlogged(self) -> None: """ Makes the table unlogged. :return: CreateQueryBuilder. """ self._unlogged = True @builder def with_system_versioning(self) -> None: """ Adds system versioning. :return: CreateQueryBuilder. """ self._with_system_versioning = True @builder def columns(self, *columns: str | tuple[str, str] | Column) -> None: """ Adds the columns. :param columns: Type: Union[str, TypedTuple[str, str], Column] A list of columns. :raises AttributeError: If the table is an as_select table. :return: CreateQueryBuilder. """ if self._as_select: raise AttributeError("'Query' object already has attribute as_select") for column in columns: if isinstance(column, str): column = Column(column) elif isinstance(column, tuple): column = Column(column_name=column[0], column_type=column[1]) self._columns.append(column) @builder def period_for(self, name, start_column: str | Column, end_column: str | Column) -> None: """ Adds a PERIOD FOR clause. :param name: The period name. :param start_column: The column that starts the period. :param end_column: The column that ends the period. :return: CreateQueryBuilder. """ self._period_fors.append(PeriodFor(name, start_column, end_column)) @builder def unique(self, *columns: str | Column) -> None: """ Adds a UNIQUE constraint. :param columns: Type: Union[str, Column] A list of columns. :return: CreateQueryBuilder. """ self._uniques.append(self._prepare_columns_input(columns)) @builder def primary_key(self, *columns: str | Column) -> None: """ Adds a primary key constraint. :param columns: Type: Union[str, Column] A list of columns. :raises AttributeError: If the primary key is already defined. :return: CreateQueryBuilder. """ if self._primary_key: raise AttributeError("'Query' object already has attribute primary_key") self._primary_key = self._prepare_columns_input(columns) @builder def foreign_key( self, columns: list[str | Column], reference_table: str | Table, reference_columns: list[str | Column], on_delete: ReferenceOption = None, on_update: ReferenceOption = None, ) -> None: """ Adds a foreign key constraint. :param columns: Type: List[Union[str, Column]] A list of foreign key columns. :param reference_table: Type: Union[str, Table] The parent table name. :param reference_columns: Type: List[Union[str, Column]] Parent key columns. :param on_delete: Type: ReferenceOption Delete action. :param on_update: Type: ReferenceOption Update option. :raises AttributeError: If the foreign key is already defined. :return: CreateQueryBuilder. """ if self._foreign_key: raise AttributeError("'Query' object already has attribute foreign_key") self._foreign_key = self._prepare_columns_input(columns) self._foreign_key_reference_table = reference_table self._foreign_key_reference = self._prepare_columns_input(reference_columns) self._foreign_key_on_delete = on_delete self._foreign_key_on_update = on_update @builder def as_select(self, query_builder: QueryBuilder) -> None: """ Creates the table from a select statement. :param query_builder: The query. :raises AttributeError: If columns have been defined for the table. :return: CreateQueryBuilder. """ if self._columns: raise AttributeError("'Query' object already has attribute columns") if not isinstance(query_builder, QueryBuilder): raise TypeError("Expected 'item' to be instance of QueryBuilder") self._as_select = query_builder @builder def if_not_exists(self) -> None: self._if_not_exists = True def get_sql(self, **kwargs: Any) -> str: """ Gets the sql statement string. :return: The create table statement. :rtype: str """ self._set_kwargs_defaults(kwargs) if not self._create_table: return "" if not self._columns and not self._as_select: return "" create_table = self._create_table_sql(**kwargs) if self._as_select: return create_table + self._as_select_sql(**kwargs) body = self._body_sql(**kwargs) table_options = self._table_options_sql(**kwargs) return f"{create_table} ({body}){table_options}" def _create_table_sql(self, **kwargs: Any) -> str: table_type = '' if self._temporary: table_type = 'TEMPORARY ' elif self._unlogged: table_type = 'UNLOGGED ' if_not_exists = '' if self._if_not_exists: if_not_exists = 'IF NOT EXISTS ' return "CREATE {table_type}TABLE {if_not_exists}{table}".format( table_type=table_type, if_not_exists=if_not_exists, table=self._create_table.get_sql(**kwargs), ) def _table_options_sql(self, **kwargs) -> str: table_options = "" if self._with_system_versioning: table_options += ' WITH SYSTEM VERSIONING' return table_options def _column_clauses(self, **kwargs) -> list[str]: return [column.get_sql(**kwargs) for column in self._columns] def _period_for_clauses(self, **kwargs) -> list[str]: return [period_for.get_sql(**kwargs) for period_for in self._period_fors] def _unique_key_clauses(self, **kwargs) -> list[str]: return [ "UNIQUE ({unique})".format(unique=",".join(column.get_name_sql(**kwargs) for column in unique)) for unique in self._uniques ] def _primary_key_clause(self, **kwargs) -> str: return "PRIMARY KEY ({columns})".format( columns=",".join(column.get_name_sql(**kwargs) for column in self._primary_key) ) def _foreign_key_clause(self, **kwargs) -> str: clause = "FOREIGN KEY ({columns}) REFERENCES {table_name} ({reference_columns})".format( columns=",".join(column.get_name_sql(**kwargs) for column in self._foreign_key), table_name=self._foreign_key_reference_table.get_sql(**kwargs), reference_columns=",".join(column.get_name_sql(**kwargs) for column in self._foreign_key_reference), ) if self._foreign_key_on_delete: clause += " ON DELETE " + self._foreign_key_on_delete.value if self._foreign_key_on_update: clause += " ON UPDATE " + self._foreign_key_on_update.value return clause def _body_sql(self, **kwargs) -> str: clauses = self._column_clauses(**kwargs) clauses += self._period_for_clauses(**kwargs) clauses += self._unique_key_clauses(**kwargs) if self._primary_key: clauses.append(self._primary_key_clause(**kwargs)) if self._foreign_key: clauses.append(self._foreign_key_clause(**kwargs)) return ",".join(clauses) def _as_select_sql(self, **kwargs: Any) -> str: return " AS ({query})".format( query=self._as_select.get_sql(**kwargs), ) def _prepare_columns_input(self, columns: list[str | Column]) -> list[Column]: return [(column if isinstance(column, Column) else Column(column)) for column in columns] def __str__(self) -> str: return self.get_sql() def __repr__(self) -> str: return self.__str__() class CreateIndexBuilder: def __init__(self) -> None: self._index = None self._columns = [] self._table = None self._wheres = None self._is_unique = False self._if_not_exists = False @builder def create_index(self, index: str | Index) -> None: self._index = index @builder def columns(self, *columns: str | tuple[str, str] | Column) -> None: for column in columns: if isinstance(column, str): column = Column(column) elif isinstance(column, tuple): column = Column(column_name=column[0], column_type=column[1]) self._columns.append(column) @builder def on(self, table: Table | str) -> None: self._table = table @builder def where(self, criterion: Term | EmptyCriterion) -> None: """ Partial index where clause. """ if self._wheres: self._wheres &= criterion else: self._wheres = criterion @builder def unique(self) -> None: self._is_unique = True @builder def if_not_exists(self) -> None: self._if_not_exists = True def get_sql(self) -> str: if not self._columns or len(self._columns) == 0: raise AttributeError("Cannot create index without columns") if not self._table: raise AttributeError("Cannot create index without table") columns_str = ", ".join([c.name for c in self._columns]) unique_str = "UNIQUE" if self._is_unique else "" if_not_exists_str = "IF NOT EXISTS" if self._if_not_exists else "" base_sql = f"CREATE {unique_str} INDEX {if_not_exists_str} {self._index} ON {self._table}({columns_str})" if self._wheres: base_sql += f" WHERE {self._wheres}" return base_sql.replace(" ", " ") def __str__(self) -> str: return self.get_sql() def __repr__(self) -> str: return self.__str__() class DropQueryBuilder: """ Query builder used to build DROP queries. """ QUOTE_CHAR = '"' SECONDARY_QUOTE_CHAR = "'" ALIAS_QUOTE_CHAR = None QUERY_CLS = Query def __init__(self, dialect: Dialects | None = None) -> None: self._drop_target_kind = None self._drop_target: Database | Table | str = "" self._if_exists = None self.dialect = dialect def _set_kwargs_defaults(self, kwargs: dict) -> None: kwargs.setdefault("quote_char", self.QUOTE_CHAR) kwargs.setdefault("secondary_quote_char", self.SECONDARY_QUOTE_CHAR) kwargs.setdefault("dialect", self.dialect) @builder def drop_database(self, database: Database | str) -> None: target = database if isinstance(database, Database) else Database(database) self._set_target('DATABASE', target) @builder def drop_table(self, table: Table | str) -> None: target = table if isinstance(table, Table) else Table(table) self._set_target('TABLE', target) @builder def drop_user(self, user: str) -> None: self._set_target('USER', user) @builder def drop_view(self, view: str) -> None: self._set_target('VIEW', view) @builder def drop_index(self, index: str) -> None: self._set_target('INDEX', index) @builder def if_exists(self) -> None: self._if_exists = True def _set_target(self, kind: str, target: Database | Table | str) -> None: if self._drop_target: raise AttributeError("'DropQuery' object already has attribute drop_target") self._drop_target_kind = kind self._drop_target = target def get_sql(self, **kwargs: Any) -> str: self._set_kwargs_defaults(kwargs) if_exists = 'IF EXISTS ' if self._if_exists else '' target_name: str = "" if isinstance(self._drop_target, (Database, Table)): target_name = self._drop_target.get_sql(**kwargs) else: target_name = format_quotes(self._drop_target, self.QUOTE_CHAR) return "DROP {kind} {if_exists}{name}".format( kind=self._drop_target_kind, if_exists=if_exists, name=target_name ) def __str__(self) -> str: return self.get_sql() def __repr__(self) -> str: return self.__str__()