sqlgen.repository.impl package

Submodules

sqlgen.repository.impl.async_constrained module

class sqlgen.repository.impl.async_constrained.AsyncConstrainedRepository(session: AsyncSession, **bound_object_ids)

Bases: AsyncRepository, ConstrainedRepository, Generic

bound_model: type[DeclarativeBase]

bound_models: list[type[DeclarativeBase]]

async create(*, safe_constraints: list[Constraint] = None, **properties) → T

override of AsyncRepository create, that will create an instance of the repository model. It will also set the foreign_key value automatically if bound_model is directly accessible, otherwise it will check for the foreign_model link to the bound_model before creating object. E.G we have three model: Project -> Host -> Webserver - if host_id is not set, an exc is raised (not if safe is set to true, but SQLAlchemy will surely if foreign_key is not nullable) - if host_id is set, a check is done for Host.project_id == self.bound_model_id (Project_id set at repo init)

Parameters:

• properties – the properties of the created model (don’t include foreign_key if directly bound, otherwise needed)

• safe_constraints – a list of constraint that are considered safe

Raises:

• BoundObjectLinkNotSafe – if the foreign_key value is not bound to the bound_model (e.g host is bound to another project)

• ConstraintNotSafe – if a non-direct constraint is not safe. e.g. Request with Bound on Webserver and Project but webserver_bound is not related to the Project_bound ( Webserver.host.project == 2 but bound_project == 1 )

• ForeignKeyNotSpecified – if no foreign_key parameters are present to create the model, bound_model relation is not direct and no other constraint in the chain are set with value

Returns:

the created object

get_constraints_for(model: type[DeclarativeBase]) → list[Constraint]

build the constraints concerning the given model (== exclude those bounding to the model)

Parameters:

model – the model to build constraint for

Returns:

a list of constraint based on those present on this repository without those linked to the model

get_repository_for(model: type[DeclarativeBase]) → AsyncConstrainedRepository

get an AsyncObjectBoundRepository for the given model (used internally to check for bound model relation on create)

Parameters:

model – the model to generate a repository for

Returns:

an AsyncObjectBoundRepository bounded to the same bound_object as this repository

async handle_constraint_not_safe(constraint: Constraint, constraints_to_test: list[Constraint], safe_constraints: list[Constraint], **properties)

exception handler for ConstraintNotSafe. check that all the “constraints to test” validate the given “constraint”.

Parameters:

• constraint – the constraint to validate

• constraints_to_test – the list of constraints to validate to ensure this constraint is safe

• safe_constraints – a list of constraints that are considered as safe and won’t be rechecked

• properties – the properties of the object to create

Returns:

the created object

async handle_direct_constraint_not_safe(model: type[DeclarativeBase], foreign_key_value: Any, constraint: Constraint, safe_constraints: list[Constraint], **properties)

exception handler for BoundObjectLinkNotSafe. check that the link to bound_model is respected using model instance identified bu foreign_key value

Parameters:

• model – the model to validate that its instance respect the bound object links

• foreign_key_value – the id of the model instance to validate

• constraint – the constraint triggering this exception (to add to safe_constraint if safe)

• safe_constraints – a list of constraints that are considered as safe and won’t be rechecked

• properties – the properties of the object to create

Returns:

the created object

statement_generator: ConstrainedStatementGenerator

sqlgen.repository.impl.async_object_bound module

class sqlgen.repository.impl.async_object_bound.AsyncObjectBoundRepository(session: AsyncSession, bound_object_id: PK)

Bases: AsyncRepository, ObjectBoundRepository, Generic

async create(safe: bool = False, **properties) → T

override of AsyncRepository create, that will create an instance of the repository model. It will also set the foreign_key value automatically if bound_model is directly accessible, otherwise it will check for the foreign_model link to the bound_model before creating object. E.G we have three model: Project -> Host -> Webserver - if host_id is not set, an exc is raised (not if safe is set to true, but SQLAlchemy will surely if foreign_key is not nullable) - if host_id is set, a check is done for Host.project_id == self.bound_model_id (Project_id set at repo init)

Parameters:

• properties – the properties of the created model (don’t include foreign_key if directly bound, otherwise needed)

• safe – a special parameter for forcing the generation even if the bound_model relation is not respected

Raises:

• BoundObjectLinkNotSafe – if the foreign_key value is not bound to the bound_model (e.g host is bound to another project)

• ForeignKeyNotSpecified – if no foreign_key parameters is present to create the model and bound_model relation is not direct (and safe=True is not set)

Returns:

the created object

get_repository_for(model: DeclarativeBase) → AsyncObjectBoundRepository

get an AsyncObjectBoundRepository for the given model (used internally to check for bound model relation on create)

Parameters:

model – the model to generate a repository for

Returns:

an AsyncObjectBoundRepository bounded to the same bound_object as this repository

sqlgen.repository.impl.asynchronous module

class sqlgen.repository.impl.asynchronous.AsyncRepository(session: AsyncSession, *args, **kwargs)

Bases: DatabaseRepository, Generic

async create(**properties) → T

generic function to create a db object

Parameters:

properties – the properties of the model

Returns:

the created object

async get_all() → Sequence

generic function to get all object of given model

Returns:

the list of instances of given model

async get_all_by(*args, options: list[ExecutableOption] = None, **kwargs) → Sequence

generic function to get all object of given model

Parameters:

• args – filter arguments for sqlalchemy filter function

• options – optional options for the query (mostly ofr joinedloads)

• kwargs – arguments to filter the model to return

Returns:

the list of instances of given model

async get_by(*args, options: list[ExecutableOption] = None, load_all: bool = False, property_name: str = None, **kwargs) → T | None

execute a sql select for the given parameters and return the object if found.

Parameters:

• args – filter arguments for sqlalchemy filter function

• options – optional options for the query (mostly ofr joinedloads)

• kwargs – arguments to filter the model to return

• load_all – load all relationship on the model when doing the query

• property_name – get a property instead of the object

Returns:

a SQL Select statement with specified options, filter, …

async get_by_id(object_id, *args, options: list[ExecutableOption] = None, load_all: bool = False, property_name: str = None, **kwargs) → T | None

get an object by its ID and return the object if found. (additional filters can be added like get_by)

Parameters:

• object_id – the id of the object to get

• args – filter arguments for sqlalchemy filter function

• options – optional options for the query (mostly ofr joinedloads)

• kwargs – arguments to filter the model to return

• load_all – load all relationship on the model when doing the query

• property_name – get a property instead of the object

Returns:

a SQL Select statement with specified options, filter, …

async restore()

restore python object state to the latest save on the DB (rollback internally)

async save()

Save the state of updated objects in DB (do a commit under the hood)

async synchronize()

synchronize the state of the DB and the python ones

async take_by(*args, options: list[ExecutableOption] = None, load_all: bool = False, property_name: str = None, **kwargs) → T

execute a sql select for the given parameters and return the object otherwise raise NoResultFound.

Parameters:

• args – filter arguments for sqlalchemy filter function

• options – optional options for the query (mostly ofr joinedloads)

• kwargs – arguments to filter the model to return

• load_all – load all relationship on the model when doing the query

• property_name – get a property instead of the object

Raises:

NoResultFound – if the query return no result

Returns:

a SQL Select statement with specified options, filter, …

async take_by_id(object_id, *args, options: list[ExecutableOption] = None, load_all: bool = False, property_name: str = None, **kwargs) → T

get an object by its ID and return the object if found otherwise raise NoResultFound. (additional filters can be added like take_by)

Parameters:

• object_id – the id of the object to get

• args – filter arguments for sqlalchemy filter function

• options – optional options for the query (mostly ofr joinedloads)

• kwargs – arguments to filter the model to return

• load_all – load all relationship on the model when doing the query

• property_name – get a property instead of the object

Returns:

a SQL Select statement with specified options, filter, …

async update(obj: DeclarativeBase | Any, save: bool = False, **properties)

update a DB object and synchronize the change with the DC

Parameters:

• obj – the object to update, or it’s ID

• save – should we commit the changes? (default to false, flush but no commit)

• properties – the properties to update on the object

sqlgen.repository.impl.synchronous module

class sqlgen.repository.impl.synchronous.SynchronousRepository(session: Session, *args, **kwargs)

Bases: DatabaseRepository, Generic

create(**properties) → T

generic function to create a db object

Parameters:

properties – the properties of the model

Returns:

the created object

get_all() → Sequence

generic function to get all object of given model

Returns:

the list of instances of given model

get_all_by(*args, options: list[ExecutableOption] = None, **kwargs) → Sequence

generic function to get all object of given model

Parameters:

• args – filter arguments for sqlalchemy filter function

• options – optional options for the query (mostly ofr joinedloads)

• kwargs – arguments to filter the model to return

Returns:

the list of instances of given model

get_by(*args, options: list[ExecutableOption] = None, load_all: bool = False, property_name: str = None, **kwargs) → T | None

execute a sql select for the given parameters and return the object if found.

Parameters:

• args – filter arguments for sqlalchemy filter function

• options – optional options for the query (mostly ofr joinedloads)

• kwargs – arguments to filter the model to return

• load_all – load all relationship on the model when doing the query

• property_name – get a property instead of the object

Returns:

a SQL Select statement with specified options, filter, …

get_by_id(object_id, *args, options: list[ExecutableOption] = None, load_all: bool = False, property_name: str = None, **kwargs) → T | None

get an object by its ID and return the object if found. (additional filters can be added like get_by)

Parameters:

• object_id – the id of the object to get

• args – filter arguments for sqlalchemy filter function

• options – optional options for the query (mostly ofr joinedloads)

• kwargs – arguments to filter the model to return

• load_all – load all relationship on the model when doing the query

• property_name – get a property instead of the object

Returns:

a SQL Select statement with specified options, filter, …

restore()

restore python object state to the latest save on the DB (rollback internally)

save()

Save the state of updated objects in DB (do a commit under the hood)

synchronize()

synchronize the state of the DB and the python ones

take_by(*args, options: list[ExecutableOption] = None, load_all: bool = False, property_name: str = None, **kwargs) → T

execute a sql select for the given parameters and return the object otherwise raise NoResultFound.

Parameters:

• args – filter arguments for sqlalchemy filter function

• options – optional options for the query (mostly ofr joinedloads)

• kwargs – arguments to filter the model to return

• load_all – load all relationship on the model when doing the query

• property_name – get a property instead of the object

Raises:

NoResultFound – if the query return no result

Returns:

a SQL Select statement with specified options, filter, …

take_by_id(object_id, *args, options: list[ExecutableOption] = None, load_all: bool = False, property_name: str = None, **kwargs) → T

get an object by its ID and return the object if found otherwise raise NoResultFound. (additional filters can be added like take_by)

Parameters:

• object_id – the id of the object to get

• args – filter arguments for sqlalchemy filter function

• options – optional options for the query (mostly ofr joinedloads)

• kwargs – arguments to filter the model to return

• load_all – load all relationship on the model when doing the query

• property_name – get a property instead of the object

Returns:

a SQL Select statement with specified options, filter, …

update(obj: DeclarativeBase | Any, save: bool = False, **properties)

update a DB object and synchronize the change with the DC

Parameters:

• obj – the object to update, or it’s ID

• save – should we commit the changes? (default to false, flush but no commit)

• properties – the properties to update on the object

sqlgen.repository.impl.synchronous_constrained module

class sqlgen.repository.impl.synchronous_constrained.SynchronousConstrainedRepository(session: Session, **bound_object_ids)

Bases: SynchronousRepository, ConstrainedRepository, Generic

bound_model: type[DeclarativeBase]

bound_models: list[type[DeclarativeBase]]

create(*, safe_constraints: list[Constraint] = None, **properties) → T

override of SynchronousRepository create, that will create an instance of the repository model. It will also set the foreign_key value automatically if bound_model is directly accessible, otherwise it will check for the foreign_model link to the bound_model before creating object. E.G we have three model: Project -> Host -> Webserver - if host_id is not set, an exc is raised (not if safe is set to true, but SQLAlchemy will surely if foreign_key is not nullable) - if host_id is set, a check is done for Host.project_id == self.bound_model_id (Project_id set at repo init)

Parameters:

• properties – the properties of the created model (don’t include foreign_key if directly bound, otherwise needed)

• safe_constraints – a list of constraint that are considered safe

Raises:

• BoundObjectLinkNotSafe – if the foreign_key value is not bound to the bound_model (e.g host is bound to another project)

• ConstraintNotSafe – if a non-direct constraint is not safe. e.g. Request with Bound on Webserver and Project but webserver_bound is not related to the Project_bound ( Webserver.host.project == 2 but bound_project == 1 )

• ForeignKeyNotSpecified – if no foreign_key parameters are present to create the model, bound_model relation is not direct and no other constraint in the chain are set with value

Returns:

the created object

get_constraints_for(model: type[DeclarativeBase]) → list[Constraint]

build the constraints concerning the given model (== exclude those bounding to the model)

Parameters:

model – the model to build constraint for

Returns:

a list of constraint based on those present on this repository without those linked to the model

get_repository_for(model: type[DeclarativeBase]) → SynchronousConstrainedRepository

get an SynchronousObjectBoundRepository for the given model (used internally to check for bound model relation on create)

Parameters:

model – the model to generate a repository for

Returns:

an SynchronousObjectBoundRepository bounded to the same bound_object as this repository

handle_constraint_not_safe(constraint: Constraint, constraints_to_test: list[Constraint], safe_constraints: list[Constraint], **properties)

exception handler for ConstraintNotSafe. check that all the “constraints to test” validate the given “constraint”.

Parameters:

• constraint – the constraint to validate

• constraints_to_test – the list of constraints to validate to ensure this constraint is safe

• safe_constraints – a list of constraints that are considered as safe and won’t be rechecked

• properties – the properties of the object to create

Returns:

the created object

handle_direct_constraint_not_safe(model: type[DeclarativeBase], foreign_key_value: Any, constraint: Constraint, safe_constraints: list[Constraint], **properties)

exception handler for BoundObjectLinkNotSafe. check that the link to bound_model is respected using model instance identified bu foreign_key value

Parameters:

• model – the model to validate that its instance respect the bound object links

• foreign_key_value – the id of the model instance to validate

• constraint – the constraint triggering this exception (to add to safe_constraint if safe)

• safe_constraints – a list of constraints that are considered as safe and won’t be rechecked

• properties – the properties of the object to create

Returns:

the created object

statement_generator: ConstrainedStatementGenerator

Module contents