sqlgen package¶
Subpackages¶
- sqlgen.repository package
- sqlgen.statement_executor package
- Submodules
- sqlgen.statement_executor.asynchronous module
AsyncStatementExecutor
AsyncStatementExecutor.all()
AsyncStatementExecutor.execute_statement()
AsyncStatementExecutor.one()
AsyncStatementExecutor.one_or_none()
AsyncStatementExecutor.restore()
AsyncStatementExecutor.save()
AsyncStatementExecutor.scalars()
AsyncStatementExecutor.store()
AsyncStatementExecutor.synchronize()
- sqlgen.statement_executor.synchronous module
SynchronousStatementExecutor
SynchronousStatementExecutor.all()
SynchronousStatementExecutor.execute_statement()
SynchronousStatementExecutor.one()
SynchronousStatementExecutor.one_or_none()
SynchronousStatementExecutor.restore()
SynchronousStatementExecutor.save()
SynchronousStatementExecutor.scalars()
SynchronousStatementExecutor.store()
SynchronousStatementExecutor.synchronize()
- Module contents
- sqlgen.statement_generator package
Submodules¶
sqlgen.exc module¶
- exception sqlgen.exc.BoundObjectLinkNotSafe(foreign_key_value, model: type[DeclarativeBase])¶
Bases:
SQLGenException
- exception sqlgen.exc.ConstraintNotSafe(constraint: Constraint, matching_constraints: list[Constraint])¶
Bases:
SQLGenException
- exception sqlgen.exc.ConstraintUninitialized(constraint: Constraint)¶
Bases:
SQLGenException
- exception sqlgen.exc.DirectConstraintNotSafe(constraint: Constraint, foreign_key_value, model: type[DeclarativeBase])¶
Bases:
SQLGenException
- exception sqlgen.exc.ForeignKeyNotSpecified(foreign_key: str)¶
Bases:
SQLGenException
- exception sqlgen.exc.MissingKeywordArgument(argument_name: str)¶
Bases:
SQLGenException
- exception sqlgen.exc.NoBoundModelDefined¶
Bases:
SQLGenException
Exception raised when trying to generate a ConstrainedRepository without constraints. Use DatabaseRepository instead
- exception sqlgen.exc.SQLGenException¶
Bases:
Exception
Base Exception Class for all sqlgen Exception
sqlgen.helpers module¶
- sqlgen.helpers.make_async_constrained_repository_class_for(model: type[T], bound_model: type[DeclarativeBase], bound_models: list[type[DeclarativeBase]]) type[AsyncConstrainedRepository] ¶
- sqlgen.helpers.make_async_object_bound_repository_class_for(model: type[T], bound_model: type[DeclarativeBase]) type[AsyncObjectBoundRepository] ¶
- sqlgen.helpers.make_async_repository_class_for(model: type[T]) type[AsyncRepository] ¶
sqlgen.joins module¶
- class sqlgen.joins.Constraint(joined_column: sqlalchemy.sql.schema.Column, joined_model: type[sqlalchemy.orm.decl_api.DeclarativeBase], joins: list[sqlalchemy.orm.attributes.InstrumentedAttribute] = <factory>, _bound_object_id: Any = <object object at 0x7cd3dcd11f90>)¶
Bases:
object
- property bound_object_id¶
- joined_column: Column¶
- joined_model: type[DeclarativeBase]¶
- joins: list[InstrumentedAttribute]¶
- set_bound_object_id(bound_object_id: Any)¶
- exception sqlgen.joins.NoValidJoins(source, destination, previous_joins)¶
Bases:
ValueError
- sqlgen.joins.resolve_model_joins(model_source: type[S], model_destination: type[D]) Constraint ¶
Resolve the joins needed to access the model_destination from the model_source
- Parameters:
model_source – the source of the joins to do
model_destination – the model to join
- Raises:
ValueError – if no chain of joins links to the model_to_join
- Returns:
a ordered list of relation attributes to access the model_destination from the model_source
sqlgen.utils module¶
- sqlgen.utils.index(__function: Callable[[T], bool], __sequence: Sequence) int ¶
Custom index function that takes a filtering function and a sequence the Sequence must stay the same after multiple iteration or an exception will be raised This function replace [].index() for SQLAlchemy table that implement custom equality behaviour [x].index(val) use x == val. but SQLAlchmey will try to generate a DDL stuff we don’t want that, we want to compare identity instead. hence the for loop with the “is” in the place of an “==” as we are comparing the same object
- Parameters:
__function – a function to find the object to get the index
__sequence – the sequence to search the item into
- Returns:
the index of the object in the sequence
- sqlgen.utils.take(__function: Callable[[T], bool], __iterable: Iterable) T ¶
take the object matching __function from __iterable :param __function: the function to use when searching the object. it should return true
if the object passed as arguments is the one searched for
- Parameters:
__iterable – the iterable to search into
- Raises:
KeyError – either no object were found or too many
- Returns:
the found object