From 17b4ce4a4dc1404ac5120134dca6f6a40ab9f937 Mon Sep 17 00:00:00 2001 From: Pierce Freeman Date: Fri, 20 Dec 2024 23:22:11 -0800 Subject: [PATCH 1/2] Expand special case select typehints to 10 attributes --- iceaxe/queries.py | 416 +++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 396 insertions(+), 20 deletions(-) diff --git a/iceaxe/queries.py b/iceaxe/queries.py index 8cc0eee..44fd742 100644 --- a/iceaxe/queries.py +++ b/iceaxe/queries.py @@ -46,6 +46,13 @@ T = TypeVar("T", bound=SUPPORTED_SELECTS) T2 = TypeVar("T2", bound=SUPPORTED_SELECTS) T3 = TypeVar("T3", bound=SUPPORTED_SELECTS) +T4 = TypeVar("T4", bound=SUPPORTED_SELECTS) +T5 = TypeVar("T5", bound=SUPPORTED_SELECTS) +T6 = TypeVar("T6", bound=SUPPORTED_SELECTS) +T7 = TypeVar("T7", bound=SUPPORTED_SELECTS) +T8 = TypeVar("T8", bound=SUPPORTED_SELECTS) +T9 = TypeVar("T9", bound=SUPPORTED_SELECTS) +T10 = TypeVar("T10", bound=SUPPORTED_SELECTS) Ts = TypeVarTuple("Ts") @@ -157,26 +164,132 @@ def select(self, fields: T | Type[T]) -> QueryBuilder[T, Literal["SELECT"]]: ... @overload def select( - self, fields: tuple[T | Type[T]] + self, + fields: tuple[T | Type[T]], ) -> QueryBuilder[tuple[T], Literal["SELECT"]]: ... @overload def select( - self, fields: tuple[T | Type[T], T2 | Type[T2]] + self, + fields: tuple[T | Type[T], T2 | Type[T2]], ) -> QueryBuilder[tuple[T, T2], Literal["SELECT"]]: ... @overload def select( - # TypeVarTuples only match the typing of one-or-more elements, so we also - # need a overloaded signature for 3 elements. self, fields: tuple[T | Type[T], T2 | Type[T2], T3 | Type[T3]], ) -> QueryBuilder[tuple[T, T2, T3], Literal["SELECT"]]: ... @overload def select( - self, fields: tuple[T | Type[T], T2 | Type[T2], T3 | Type[T3], *Ts] - ) -> QueryBuilder[tuple[T, T2, T3, *Ts], Literal["SELECT"]]: ... + self, + fields: tuple[T | Type[T], T2 | Type[T2], T3 | Type[T3], T4 | Type[T4]], + ) -> QueryBuilder[tuple[T, T2, T3, T4], Literal["SELECT"]]: ... + + @overload + def select( + self, + fields: tuple[ + T | Type[T], T2 | Type[T2], T3 | Type[T3], T4 | Type[T4], T5 | Type[T5] + ], + ) -> QueryBuilder[tuple[T, T2, T3, T4, T5], Literal["SELECT"]]: ... + + @overload + def select( + self, + fields: tuple[ + T | Type[T], + T2 | Type[T2], + T3 | Type[T3], + T4 | Type[T4], + T5 | Type[T5], + T6 | Type[T6], + ], + ) -> QueryBuilder[tuple[T, T2, T3, T4, T5, T6], Literal["SELECT"]]: ... + + @overload + def select( + self, + fields: tuple[ + T | Type[T], + T2 | Type[T2], + T3 | Type[T3], + T4 | Type[T4], + T5 | Type[T5], + T6 | Type[T6], + T7 | Type[T7], + ], + ) -> QueryBuilder[tuple[T, T2, T3, T4, T5, T6, T7], Literal["SELECT"]]: ... + + @overload + def select( + self, + fields: tuple[ + T | Type[T], + T2 | Type[T2], + T3 | Type[T3], + T4 | Type[T4], + T5 | Type[T5], + T6 | Type[T6], + T7 | Type[T7], + T8 | Type[T8], + ], + ) -> QueryBuilder[tuple[T, T2, T3, T4, T5, T6, T7, T8], Literal["SELECT"]]: ... + + @overload + def select( + self, + fields: tuple[ + T | Type[T], + T2 | Type[T2], + T3 | Type[T3], + T4 | Type[T4], + T5 | Type[T5], + T6 | Type[T6], + T7 | Type[T7], + T8 | Type[T8], + T9 | Type[T9], + ], + ) -> QueryBuilder[tuple[T, T2, T3, T4, T5, T6, T7, T8, T9], Literal["SELECT"]]: ... + + @overload + def select( + self, + fields: tuple[ + T | Type[T], + T2 | Type[T2], + T3 | Type[T3], + T4 | Type[T4], + T5 | Type[T5], + T6 | Type[T6], + T7 | Type[T7], + T8 | Type[T8], + T9 | Type[T9], + T10 | Type[T10], + ], + ) -> QueryBuilder[ + tuple[T, T2, T3, T4, T5, T6, T7, T8, T9, T10], Literal["SELECT"] + ]: ... + + @overload + def select( + self, + fields: tuple[ + T | Type[T], + T2 | Type[T2], + T3 | Type[T3], + T4 | Type[T4], + T5 | Type[T5], + T6 | Type[T6], + T7 | Type[T7], + T8 | Type[T8], + T9 | Type[T9], + T10 | Type[T10], + *Ts, + ], + ) -> QueryBuilder[ + tuple[T, T2, T3, T4, T5, T6, T7, T8, T9, T10, *Ts], Literal["SELECT"] + ]: ... @allow_branching def select( @@ -187,14 +300,89 @@ def select( | tuple[T | Type[T]] | tuple[T | Type[T], T2 | Type[T2]] | tuple[T | Type[T], T2 | Type[T2], T3 | Type[T3]] - | tuple[T | Type[T], T2 | Type[T2], T3 | Type[T3], *Ts] + | tuple[T | Type[T], T2 | Type[T2], T3 | Type[T3], T4 | Type[T4]] + | tuple[ + T | Type[T], T2 | Type[T2], T3 | Type[T3], T4 | Type[T4], T5 | Type[T5] + ] + | tuple[ + T | Type[T], + T2 | Type[T2], + T3 | Type[T3], + T4 | Type[T4], + T5 | Type[T5], + T6 | Type[T6], + ] + | tuple[ + T | Type[T], + T2 | Type[T2], + T3 | Type[T3], + T4 | Type[T4], + T5 | Type[T5], + T6 | Type[T6], + T7 | Type[T7], + ] + | tuple[ + T | Type[T], + T2 | Type[T2], + T3 | Type[T3], + T4 | Type[T4], + T5 | Type[T5], + T6 | Type[T6], + T7 | Type[T7], + T8 | Type[T8], + ] + | tuple[ + T | Type[T], + T2 | Type[T2], + T3 | Type[T3], + T4 | Type[T4], + T5 | Type[T5], + T6 | Type[T6], + T7 | Type[T7], + T8 | Type[T8], + T9 | Type[T9], + ] + | tuple[ + T | Type[T], + T2 | Type[T2], + T3 | Type[T3], + T4 | Type[T4], + T5 | Type[T5], + T6 | Type[T6], + T7 | Type[T7], + T8 | Type[T8], + T9 | Type[T9], + T10 | Type[T10], + ] + | tuple[ + T | Type[T], + T2 | Type[T2], + T3 | Type[T3], + T4 | Type[T4], + T5 | Type[T5], + T6 | Type[T6], + T7 | Type[T7], + T8 | Type[T8], + T9 | Type[T9], + T10 | Type[T10], + *Ts, + ] ), ) -> ( QueryBuilder[T, Literal["SELECT"]] | QueryBuilder[tuple[T], Literal["SELECT"]] | QueryBuilder[tuple[T, T2], Literal["SELECT"]] | QueryBuilder[tuple[T, T2, T3], Literal["SELECT"]] - | QueryBuilder[tuple[T, T2, T3, *Ts], Literal["SELECT"]] + | QueryBuilder[tuple[T, T2, T3, T4], Literal["SELECT"]] + | QueryBuilder[tuple[T, T2, T3, T4, T5], Literal["SELECT"]] + | QueryBuilder[tuple[T, T2, T3, T4, T5, T6], Literal["SELECT"]] + | QueryBuilder[tuple[T, T2, T3, T4, T5, T6, T7], Literal["SELECT"]] + | QueryBuilder[tuple[T, T2, T3, T4, T5, T6, T7, T8], Literal["SELECT"]] + | QueryBuilder[tuple[T, T2, T3, T4, T5, T6, T7, T8, T9], Literal["SELECT"]] + | QueryBuilder[tuple[T, T2, T3, T4, T5, T6, T7, T8, T9, T10], Literal["SELECT"]] + | QueryBuilder[ + tuple[T, T2, T3, T4, T5, T6, T7, T8, T9, T10, *Ts], Literal["SELECT"] + ] ): """ Creates a SELECT query to fetch data from the database. @@ -240,7 +428,8 @@ def select( and not is_function_metadata(field) ): raise ValueError( - f"Invalid field type {field}. Must be:\n1. A column field\n2. A table\n3. A QueryLiteral\n4. A tuple of the above." + f"Invalid field type { + field}. Must be:\n1. A column field\n2. A table\n3. A QueryLiteral\n4. A tuple of the above." ) self._select_inner(all_fields) @@ -441,7 +630,8 @@ def join(self, table: Type[TableBase], on: bool, join_type: JoinType = "INNER"): """ if not is_comparison(on): raise ValueError( - f"Invalid join condition: {on}, should be MyTable.column == OtherTable.column" + f"Invalid join condition: { + on}, should be MyTable.column == OtherTable.column" ) table_name = QueryLiteral(table.get_table_name()) @@ -449,7 +639,8 @@ def join(self, table: Type[TableBase], on: bool, join_type: JoinType = "INNER"): comparison = QueryLiteral(on.comparison.value) on_right, _ = on.right.to_query() - join_sql = f"{join_type} JOIN {table_name} ON {on_left} {comparison} {on_right}" + join_sql = f"{join_type} JOIN {table_name} ON { + on_left} {comparison} {on_right}" self._join_clauses.append(join_sql) return self @@ -774,7 +965,8 @@ def build(self) -> tuple[str, list[Any]]: set_components = [] for column, value in self._update_values: column_token, _ = column.to_query() - set_components.append(f"{column_token} = ${len(variables) + 1}") + set_components.append(f"{column_token} = ${ + len(variables) + 1}") variables.append(value) set_clause = ", ".join(set_components) @@ -800,7 +992,8 @@ def build(self) -> tuple[str, list[Any]]: if self._group_by_fields: query += " GROUP BY " query += ", ".join( - f"{QueryIdentifier(field.root_model.get_table_name())}.{QueryIdentifier(field.key)}" + f"{QueryIdentifier(field.root_model.get_table_name())}.{ + QueryIdentifier(field.key)}" for field in self._group_by_fields ) @@ -818,9 +1011,8 @@ def build(self) -> tuple[str, list[Any]]: variables.append(having_condition.right) having_value = QueryLiteral("$" + str(len(variables))) - query += ( - f"{having_field} {having_condition.comparison.value} {having_value}" - ) + query += f"{having_field} { + having_condition.comparison.value} {having_value}" if self._order_by_clauses: query += " ORDER BY " + ", ".join(self._order_by_clauses) @@ -940,8 +1132,117 @@ def select( @overload def select( - fields: tuple[T | Type[T], T2 | Type[T2], T3 | Type[T3], *Ts], -) -> QueryBuilder[tuple[T, T2, T3, *Ts], Literal["SELECT"]]: ... + fields: tuple[T | Type[T], T2 | Type[T2], T3 | Type[T3]], +) -> QueryBuilder[tuple[T, T2, T3], Literal["SELECT"]]: ... + + +@overload +def select( + fields: tuple[T | Type[T], T2 | Type[T2], T3 | Type[T3], T4 | Type[T4]], +) -> QueryBuilder[tuple[T, T2, T3, T4], Literal["SELECT"]]: ... + + +@overload +def select( + fields: tuple[ + T | Type[T], T2 | Type[T2], T3 | Type[T3], T4 | Type[T4], T5 | Type[T5] + ], +) -> QueryBuilder[tuple[T, T2, T3, T4, T5], Literal["SELECT"]]: ... + + +@overload +def select( + fields: tuple[ + T | Type[T], + T2 | Type[T2], + T3 | Type[T3], + T4 | Type[T4], + T5 | Type[T5], + T6 | Type[T6], + ], +) -> QueryBuilder[tuple[T, T2, T3, T4, T5, T6], Literal["SELECT"]]: ... + + +@overload +def select( + fields: tuple[ + T | Type[T], + T2 | Type[T2], + T3 | Type[T3], + T4 | Type[T4], + T5 | Type[T5], + T6 | Type[T6], + T7 | Type[T7], + ], +) -> QueryBuilder[tuple[T, T2, T3, T4, T5, T6, T7], Literal["SELECT"]]: ... + + +@overload +def select( + fields: tuple[ + T | Type[T], + T2 | Type[T2], + T3 | Type[T3], + T4 | Type[T4], + T5 | Type[T5], + T6 | Type[T6], + T7 | Type[T7], + T8 | Type[T8], + ], +) -> QueryBuilder[tuple[T, T2, T3, T4, T5, T6, T7, T8], Literal["SELECT"]]: ... + + +@overload +def select( + fields: tuple[ + T | Type[T], + T2 | Type[T2], + T3 | Type[T3], + T4 | Type[T4], + T5 | Type[T5], + T6 | Type[T6], + T7 | Type[T7], + T8 | Type[T8], + T9 | Type[T9], + ], +) -> QueryBuilder[tuple[T, T2, T3, T4, T5, T6, T7, T8, T9], Literal["SELECT"]]: ... + + +@overload +def select( + fields: tuple[ + T | Type[T], + T2 | Type[T2], + T3 | Type[T3], + T4 | Type[T4], + T5 | Type[T5], + T6 | Type[T6], + T7 | Type[T7], + T8 | Type[T8], + T9 | Type[T9], + T10 | Type[T10], + ], +) -> QueryBuilder[tuple[T, T2, T3, T4, T5, T6, T7, T8, T9, T10], Literal["SELECT"]]: ... + + +@overload +def select( + fields: tuple[ + T | Type[T], + T2 | Type[T2], + T3 | Type[T3], + T4 | Type[T4], + T5 | Type[T5], + T6 | Type[T6], + T7 | Type[T7], + T8 | Type[T8], + T9 | Type[T9], + T10 | Type[T10], + *Ts, + ], +) -> QueryBuilder[ + tuple[T, T2, T3, T4, T5, T6, T7, T8, T9, T10, *Ts], Literal["SELECT"] +]: ... def select( @@ -950,13 +1251,88 @@ def select( | Type[T] | tuple[T | Type[T]] | tuple[T | Type[T], T2 | Type[T2]] - | tuple[T | Type[T], T2 | Type[T2], T3 | Type[T3], *Ts] + | tuple[T | Type[T], T2 | Type[T2], T3 | Type[T3]] + | tuple[T | Type[T], T2 | Type[T2], T3 | Type[T3], T4 | Type[T4]] + | tuple[T | Type[T], T2 | Type[T2], T3 | Type[T3], T4 | Type[T4], T5 | Type[T5]] + | tuple[ + T | Type[T], + T2 | Type[T2], + T3 | Type[T3], + T4 | Type[T4], + T5 | Type[T5], + T6 | Type[T6], + ] + | tuple[ + T | Type[T], + T2 | Type[T2], + T3 | Type[T3], + T4 | Type[T4], + T5 | Type[T5], + T6 | Type[T6], + T7 | Type[T7], + ] + | tuple[ + T | Type[T], + T2 | Type[T2], + T3 | Type[T3], + T4 | Type[T4], + T5 | Type[T5], + T6 | Type[T6], + T7 | Type[T7], + T8 | Type[T8], + ] + | tuple[ + T | Type[T], + T2 | Type[T2], + T3 | Type[T3], + T4 | Type[T4], + T5 | Type[T5], + T6 | Type[T6], + T7 | Type[T7], + T8 | Type[T8], + T9 | Type[T9], + ] + | tuple[ + T | Type[T], + T2 | Type[T2], + T3 | Type[T3], + T4 | Type[T4], + T5 | Type[T5], + T6 | Type[T6], + T7 | Type[T7], + T8 | Type[T8], + T9 | Type[T9], + T10 | Type[T10], + ] + | tuple[ + T | Type[T], + T2 | Type[T2], + T3 | Type[T3], + T4 | Type[T4], + T5 | Type[T5], + T6 | Type[T6], + T7 | Type[T7], + T8 | Type[T8], + T9 | Type[T9], + T10 | Type[T10], + *Ts, + ] ), ) -> ( QueryBuilder[T, Literal["SELECT"]] | QueryBuilder[tuple[T], Literal["SELECT"]] | QueryBuilder[tuple[T, T2], Literal["SELECT"]] - | QueryBuilder[tuple[T, T2, T3, *Ts], Literal["SELECT"]] + | QueryBuilder[tuple[T, T2, T3], Literal["SELECT"]] + | QueryBuilder[tuple[T, T2, T3, T4], Literal["SELECT"]] + | QueryBuilder[tuple[T, T2, T3, T4, T5], Literal["SELECT"]] + | QueryBuilder[tuple[T, T2, T3, T4, T5, T6], Literal["SELECT"]] + | QueryBuilder[tuple[T, T2, T3, T4, T5, T6, T7], Literal["SELECT"]] + | QueryBuilder[tuple[T, T2, T3, T4, T5, T6, T7, T8], Literal["SELECT"]] + | QueryBuilder[tuple[T, T2, T3, T4, T5, T6, T7, T8, T9], Literal["SELECT"]] + | QueryBuilder[tuple[T, T2, T3, T4, T5, T6, T7, T8, T9, T10], Literal["SELECT"]] + | QueryBuilder[ + tuple[T, T2, T3, T4, T5, T6, T7, T8, T9, T10, *Ts], Literal["SELECT"] + ] ): """ Creates a SELECT query to fetch data from the database. This is a shortcut function that creates From 3e344973af3a5d2c41e3f4a464f5fc523f2c26a2 Mon Sep 17 00:00:00 2001 From: Pierce Freeman Date: Fri, 20 Dec 2024 23:43:31 -0800 Subject: [PATCH 2/2] Fix formatted strings --- iceaxe/queries.py | 18 ++++++------------ 1 file changed, 6 insertions(+), 12 deletions(-) diff --git a/iceaxe/queries.py b/iceaxe/queries.py index 44fd742..b971fa8 100644 --- a/iceaxe/queries.py +++ b/iceaxe/queries.py @@ -428,8 +428,7 @@ def select( and not is_function_metadata(field) ): raise ValueError( - f"Invalid field type { - field}. Must be:\n1. A column field\n2. A table\n3. A QueryLiteral\n4. A tuple of the above." + f"Invalid field type {field}. Must be:\n1. A column field\n2. A table\n3. A QueryLiteral\n4. A tuple of the above." ) self._select_inner(all_fields) @@ -630,8 +629,7 @@ def join(self, table: Type[TableBase], on: bool, join_type: JoinType = "INNER"): """ if not is_comparison(on): raise ValueError( - f"Invalid join condition: { - on}, should be MyTable.column == OtherTable.column" + f"Invalid join condition: {on}, should be MyTable.column == OtherTable.column" ) table_name = QueryLiteral(table.get_table_name()) @@ -639,8 +637,7 @@ def join(self, table: Type[TableBase], on: bool, join_type: JoinType = "INNER"): comparison = QueryLiteral(on.comparison.value) on_right, _ = on.right.to_query() - join_sql = f"{join_type} JOIN {table_name} ON { - on_left} {comparison} {on_right}" + join_sql = f"{join_type} JOIN {table_name} ON {on_left} {comparison} {on_right}" self._join_clauses.append(join_sql) return self @@ -965,8 +962,7 @@ def build(self) -> tuple[str, list[Any]]: set_components = [] for column, value in self._update_values: column_token, _ = column.to_query() - set_components.append(f"{column_token} = ${ - len(variables) + 1}") + set_components.append(f"{column_token} = ${len(variables) + 1}") variables.append(value) set_clause = ", ".join(set_components) @@ -992,8 +988,7 @@ def build(self) -> tuple[str, list[Any]]: if self._group_by_fields: query += " GROUP BY " query += ", ".join( - f"{QueryIdentifier(field.root_model.get_table_name())}.{ - QueryIdentifier(field.key)}" + f"{QueryIdentifier(field.root_model.get_table_name())}.{QueryIdentifier(field.key)}" for field in self._group_by_fields ) @@ -1011,8 +1006,7 @@ def build(self) -> tuple[str, list[Any]]: variables.append(having_condition.right) having_value = QueryLiteral("$" + str(len(variables))) - query += f"{having_field} { - having_condition.comparison.value} {having_value}" + query += f"{having_field} {having_condition.comparison.value} {having_value}" if self._order_by_clauses: query += " ORDER BY " + ", ".join(self._order_by_clauses)