|
# Centralized query bindings. |
|
# A query entry should be the query name (string), then its value should be a tuple of the ORM or source model and the GraphQL model. |
|
# These are used to dynamically build all necessary queries for the GraphQL router. |
|
_query_model_bindings: Dict[str, Tuple[Type[Base], Type[BaseModelType]]] = { |
|
"cars": (models.Car, Car), |
|
"bikes": (models.Bike, Bike), |
|
} |
|
|
|
|
|
def _generate_dynamic_query_class(): |
|
def generate_bind(query_name: str, model: Base, schema: Type[BaseModelType]): |
|
""" |
|
Generate function instances per query by closure. |
|
""" |
|
def _query( |
|
self, # noqa |
|
info: strawberry.Info, |
|
_id: Annotated[ |
|
int | None, |
|
strawberry.argument( |
|
description="The primary key item of the object desired." |
|
) |
|
] = None, |
|
) -> Sequence[schema]: |
|
""" |
|
Baseline GraphQL query logic. |
|
:param self: Query class instance. |
|
:param info: The Strawberry context instance. Our custom database session is present here. |
|
:param _id: The ID of the object to query for. |
|
:return: |
|
""" |
|
if _id: |
|
# TODO: Use your logic here. This is an example of a simple SQLAlchemy database session received from the context. |
|
# Because we can't use union types, in order to consolidate fetch_one and fetch_all into one endpoint, make this a list with a single item. |
|
return [info.context["db_session"].scalar(select(model).where(model.id == _id))] |
|
else: |
|
return info.context["db_session"].scalars(select(model)).all() |
|
|
|
# Replace the function's name with the query name. |
|
_query.__name__ = query_name |
|
|
|
# Once the query function is built, return it to be assigned to the query class. |
|
return _query |
|
|
|
class Query: |
|
""" |
|
Baseline query class. Query functions will be dynamically assigned to this object. |
|
""" |
|
... |
|
|
|
# Iterate over the query-to-model bindings and assign generated bindings to the Query class. |
|
for (_query_name, (_model, _schema)) in _query_model_bindings.items(): |
|
setattr(Query, _query_name, strawberry.field(generate_bind(_query_name, _model, _schema))) |
|
|
|
# Decorate the query class, initialize/finalize the mapper, and build the strawberry schema. |
|
strawberry.type(Query) |
|
strawberry_sqlalchemy_mapper.finalize() |
|
_additional_types = list(strawberry_sqlalchemy_mapper.mapped_types.values()) |
|
return strawberry.Schema( |
|
query=Query, |
|
types=_additional_types, |
|
) |
|
|
|
|
|
# Generate a GraphQL schema with the dynamically built GraphQL query. |
|
# This would then be used in the GraphQL app or router. |
|
graphql_schema = _generate_dynamic_query_class() |