Repositories

The repositories that run a field's query against the session, in sync and async variants.

`StrawchemyAsyncRepository`

python

StrawchemyAsyncRepository(type: type[T], info: Info[Any, Any], root_aggregations: bool = False, auto_snake_case: bool = True, query_hook: QueryHook[Any] | Sequence[QueryHook[Any]] | None = None, session_getter: AsyncSessionGetter = default_session_getter, session: AnyAsyncSession | None = None, filter_statement: SelectOf[Any] | None = None, execution_options: dict[str, Any] | None = None, deterministic_ordering: bool = False, default_order_by: builtins.list[OrderByExpr] = list()) -> None

Asynchronous strawberry implementation for GraphQL data access.

This class provides asynchronous methods for querying and mutating data through GraphQL, using SQLAlchemy's asynchronous API under the hood.

Parameters:

Name	Type	Description	Default
`type`	`type[T]`	The Strawberry GraphQL type this strawberry works with	required
`info`	`Info[Any, Any]`	The GraphQL resolver info object	required
`session_getter`	`AsyncSessionGetter`	Callable to get an async database session	`default_session_getter`
`session`	`AnyAsyncSession \| None`	Optional explicit async database session to use	`None`
`filter_statement`	`SelectOf[Any] \| None`	Optional base SQLAlchemy select statement to apply to all queries	`None`
`execution_options`	`dict[str, Any] \| None`	Optional execution options for SQLAlchemy	`None`
`deterministic_ordering`	`bool`	Whether to ensure deterministic ordering of results	`False`

`auto_snake_case`

python

auto_snake_case: bool = True

`create`

python

create(data: Input[InputModel]) -> GraphQLResult[InputModel, T]

Asynchronously create a new entity.

Parameters:

Name	Type	Description	Default
`data`	`Input[InputModel]`	The input data for the new entity	required

Returns:

Type	Description
`GraphQLResult[InputModel, T]`	A GraphQLResult containing the created entity

`default_order_by`

python

default_order_by: builtins.list[OrderByExpr] = field(default_factory=list)

`delete`

python

delete(filter_input: BooleanFilterDTO | None) -> GraphQLResult[Any, T]

Asynchronously delete entities matching the given filter.

Parameters:

Name	Type	Description	Default
`filter_input`	`BooleanFilterDTO \| None`	The filter criteria to select entities to delete	required

Returns:

Type	Description
`GraphQLResult[Any, T]`	A GraphQLResult containing the deleted entities

`deterministic_ordering`

python

deterministic_ordering: bool = False

`execution_options`

python

execution_options: dict[str, Any] | None = None

`filter_statement`

python

filter_statement: SelectOf[Any] | None = None

`get_by_id`

python

get_by_id(**kwargs: Any) -> GraphQLResult[Any, T]

Asynchronously get an entity by its primary key.

Parameters:

Name	Type	Description	Default
`**kwargs`	`Any`	Primary key field names and values	`{}`

Returns:

Type	Description
`GraphQLResult[Any, T]`	A GraphQLResult containing the found entity

Raises:

Type	Description
`NoResultFound`	If no entity with the given ID exists

`get_one`

python

get_one(filter_input: BooleanFilterDTO | None = None, order_by: builtins.list[OrderByDTO] | None = None, distinct_on: builtins.list[EnumDTO] | None = None, limit: int | None = None, offset: int | None = None) -> GraphQLResult[Any, T]

Asynchronously get exactly one result matching the criteria.

Parameters:

Name	Type	Description	Default
`filter_input`	`BooleanFilterDTO \| None`	Optional filter conditions	`None`
`order_by`	`list[OrderByDTO] \| None`	Optional ordering criteria	`None`
`distinct_on`	`list[EnumDTO] \| None`	Optional fields to apply DISTINCT on	`None`
`limit`	`int \| None`	Optional maximum number of results to return	`None`
`offset`	`int \| None`	Optional number of results to skip	`None`

Returns:

Type	Description
`GraphQLResult[Any, T]`	A GraphQLResult containing the single result

Raises:

Type	Description
`NoResultFound`	If no results are found
`MultipleResultsFound`	If multiple results are found

`get_one_or_none`

python

get_one_or_none(filter_input: BooleanFilterDTO | None = None, order_by: builtins.list[OrderByDTO] | None = None, distinct_on: builtins.list[EnumDTO] | None = None, limit: int | None = None, offset: int | None = None) -> GraphQLResult[Any, T]

Asynchronously get at most one result matching the criteria or None.

Parameters:

Name	Type	Description	Default
`filter_input`	`BooleanFilterDTO \| None`	Optional filter conditions	`None`
`order_by`	`list[OrderByDTO] \| None`	Optional ordering criteria	`None`
`distinct_on`	`list[EnumDTO] \| None`	Optional fields to apply DISTINCT on	`None`
`limit`	`int \| None`	Optional maximum number of results to return	`None`
`offset`	`int \| None`	Optional number of results to skip	`None`

Returns:

Type	Description
`GraphQLResult[Any, T]`	A GraphQLResult containing the result or None

`graphql_repository`

python

graphql_repository() -> SQLAlchemyGraphQLAsyncRepository[Any]

Create and configure the underlying async SQLAlchemy GraphQL strawberry.

Returns:

Type	Description
`SQLAlchemyGraphQLAsyncRepository[Any]`	A configured SQLAlchemyGraphQLAsyncRepository instance

`info`

python

info: Info[Any, Any]

`is_async`

python

is_async = IS_ASYNC_REPOSITORY

`list`

python

list(filter_input: BooleanFilterDTO | None = None, order_by: builtins.list[OrderByDTO] | None = None, distinct_on: builtins.list[EnumDTO] | None = None, limit: int | None = None, offset: int | None = None) -> GraphQLResult[Any, T]

Asynchronously get a list of entities matching the criteria.

Parameters:

Name	Type	Description	Default
`filter_input`	`BooleanFilterDTO \| None`	Optional filter conditions	`None`
`order_by`	`list[OrderByDTO] \| None`	Optional ordering criteria	`None`
`distinct_on`	`list[EnumDTO] \| None`	Optional fields to apply DISTINCT on	`None`
`limit`	`int \| None`	Optional maximum number of results to return	`None`
`offset`	`int \| None`	Optional number of results to skip	`None`

Returns:

Type	Description
`GraphQLResult[Any, T]`	A GraphQLResult containing the list of matching entities

`query_hook`

python

query_hook: QueryHook[Any] | Sequence[QueryHook[Any]] | None = None

`root_aggregations`

python

root_aggregations: bool = False

`session`

python

session: AnyAsyncSession | None = None

`session_getter`

python

session_getter: AsyncSessionGetter = default_session_getter

`type`

python

type: type[T]

`update_by_filter`

python

update_by_filter(data: Input[InputModel], filter_input: BooleanFilterDTO) -> GraphQLResult[InputModel, T]

Asynchronously update entities matching the given filter.

Parameters:

Name	Type	Description	Default
`data`	`Input[InputModel]`	The input data containing fields to update	required
`filter_input`	`BooleanFilterDTO`	The filter criteria to select entities to update	required

Returns:

Type	Description
`GraphQLResult[InputModel, T]`	A GraphQLResult containing the updated entities

`update_by_id`

python

update_by_id(data: Input[InputModel]) -> GraphQLResult[InputModel, T]

Asynchronously update an entity by its ID.

Parameters:

Name	Type	Description	Default
`data`	`Input[InputModel]`	The input data containing the ID and fields to update	required

Returns:

Type	Description
`GraphQLResult[InputModel, T]`	A GraphQLResult containing the updated entity

Raises:

Type	Description
`NoResultFound`	If no entity with the given ID exists

`upsert`

python

upsert(data: Input[InputModel], filter_input: BooleanFilterDTO | None = None, update_fields: builtins.list[EnumDTO] | None = None, conflict_fields: EnumDTO | None = None) -> GraphQLResult[InputModel, T]

Asynchronously insert or update an entity.

Parameters:

Name	Type	Description	Default
`data`	`Input[InputModel]`	The input data for the entity	required
`filter_input`	`BooleanFilterDTO \| None`	Optional filter to find existing entity	`None`
`update_fields`	`list[EnumDTO] \| None`	Optional fields to update if entity exists	`None`
`conflict_fields`	`EnumDTO \| None`	Optional fields to detect conflicts on	`None`

Returns:

Type	Description
`GraphQLResult[InputModel, T]`	A GraphQLResult containing the upserted entity

`StrawchemySyncRepository`

source

python

StrawchemySyncRepository(type: type[T], info: Info[Any, Any], root_aggregations: bool = False, auto_snake_case: bool = True, query_hook: QueryHook[Any] | Sequence[QueryHook[Any]] | None = None, session_getter: SyncSessionGetter = default_session_getter, session: AnySyncSession | None = None, filter_statement: SelectOf[Any] | None = None, execution_options: dict[str, Any] | None = None, deterministic_ordering: bool = False, default_order_by: builtins.list[OrderByExpr] = list()) -> None

Asynchronous strawberry implementation for GraphQL data access.

This class provides asynchronous methods for querying and mutating data through GraphQL, using SQLAlchemy's asynchronous API under the hood.

Parameters:

Name	Type	Description	Default
`type`	`type[T]`	The Strawberry GraphQL type this strawberry works with	required
`info`	`Info[Any, Any]`	The GraphQL resolver info object	required
`session_getter`	`SyncSessionGetter`	Callable to get an async database session	`default_session_getter`
`session`	`AnySyncSession \| None`	Optional explicit async database session to use	`None`
`filter_statement`	`SelectOf[Any] \| None`	Optional base SQLAlchemy select statement to apply to all queries	`None`
`execution_options`	`dict[str, Any] \| None`	Optional execution options for SQLAlchemy	`None`
`deterministic_ordering`	`bool`	Whether to ensure deterministic ordering of results	`False`

`auto_snake_case`

python

auto_snake_case: bool = True

`create`

python

create(data: Input[InputModel]) -> GraphQLResult[InputModel, T]

Asynchronously create a new entity.

Parameters:

Name	Type	Description	Default
`data`	`Input[InputModel]`	The input data for the new entity	required

Returns:

Type	Description
`GraphQLResult[InputModel, T]`	A GraphQLResult containing the created entity

`default_order_by`

python

default_order_by: builtins.list[OrderByExpr] = field(default_factory=list)

`delete`

python

delete(filter_input: BooleanFilterDTO | None) -> GraphQLResult[Any, T]

Asynchronously delete entities matching the given filter.

Parameters:

Name	Type	Description	Default
`filter_input`	`BooleanFilterDTO \| None`	The filter criteria to select entities to delete	required

Returns:

Type	Description
`GraphQLResult[Any, T]`	A GraphQLResult containing the deleted entities

`deterministic_ordering`

python

deterministic_ordering: bool = False

`execution_options`

python

execution_options: dict[str, Any] | None = None

`filter_statement`

python

filter_statement: SelectOf[Any] | None = None

`get_by_id`

python

get_by_id(**kwargs: Any) -> GraphQLResult[Any, T]

Asynchronously get an entity by its primary key.

Parameters:

Name	Type	Description	Default
`**kwargs`	`Any`	Primary key field names and values	`{}`

Returns:

Type	Description
`GraphQLResult[Any, T]`	A GraphQLResult containing the found entity

Raises:

Type	Description
`NoResultFound`	If no entity with the given ID exists

`get_one`

python

get_one(filter_input: BooleanFilterDTO | None = None, order_by: builtins.list[OrderByDTO] | None = None, distinct_on: builtins.list[EnumDTO] | None = None, limit: int | None = None, offset: int | None = None) -> GraphQLResult[Any, T]

Asynchronously get exactly one result matching the criteria.

Parameters:

Name	Type	Description	Default
`filter_input`	`BooleanFilterDTO \| None`	Optional filter conditions	`None`
`order_by`	`list[OrderByDTO] \| None`	Optional ordering criteria	`None`
`distinct_on`	`list[EnumDTO] \| None`	Optional fields to apply DISTINCT on	`None`
`limit`	`int \| None`	Optional maximum number of results to return	`None`
`offset`	`int \| None`	Optional number of results to skip	`None`

Returns:

Type	Description
`GraphQLResult[Any, T]`	A GraphQLResult containing the single result

Raises:

Type	Description
`NoResultFound`	If no results are found
`MultipleResultsFound`	If multiple results are found

`get_one_or_none`

python

get_one_or_none(filter_input: BooleanFilterDTO | None = None, order_by: builtins.list[OrderByDTO] | None = None, distinct_on: builtins.list[EnumDTO] | None = None, limit: int | None = None, offset: int | None = None) -> GraphQLResult[Any, T]

Asynchronously get at most one result matching the criteria or None.

Parameters:

Name	Type	Description	Default
`filter_input`	`BooleanFilterDTO \| None`	Optional filter conditions	`None`
`order_by`	`list[OrderByDTO] \| None`	Optional ordering criteria	`None`
`distinct_on`	`list[EnumDTO] \| None`	Optional fields to apply DISTINCT on	`None`
`limit`	`int \| None`	Optional maximum number of results to return	`None`
`offset`	`int \| None`	Optional number of results to skip	`None`

Returns:

Type	Description
`GraphQLResult[Any, T]`	A GraphQLResult containing the result or None

`graphql_repository`

python

graphql_repository() -> SQLAlchemyGraphQLSyncRepository[Any]

Create and configure the underlying async SQLAlchemy GraphQL strawberry.

Returns:

Type	Description
`SQLAlchemyGraphQLSyncRepository[Any]`	A configured SQLAlchemyGraphQLAsyncRepository instance

`info`

python

info: Info[Any, Any]

`is_async`

python

is_async = IS_SYNC_REPOSITORY

`list`

python

list(filter_input: BooleanFilterDTO | None = None, order_by: builtins.list[OrderByDTO] | None = None, distinct_on: builtins.list[EnumDTO] | None = None, limit: int | None = None, offset: int | None = None) -> GraphQLResult[Any, T]

Asynchronously get a list of entities matching the criteria.

Parameters:

Name	Type	Description	Default
`filter_input`	`BooleanFilterDTO \| None`	Optional filter conditions	`None`
`order_by`	`list[OrderByDTO] \| None`	Optional ordering criteria	`None`
`distinct_on`	`list[EnumDTO] \| None`	Optional fields to apply DISTINCT on	`None`
`limit`	`int \| None`	Optional maximum number of results to return	`None`
`offset`	`int \| None`	Optional number of results to skip	`None`

Returns:

Type	Description
`GraphQLResult[Any, T]`	A GraphQLResult containing the list of matching entities

`query_hook`

python

query_hook: QueryHook[Any] | Sequence[QueryHook[Any]] | None = None

`root_aggregations`

python

root_aggregations: bool = False

`session`

python

session: AnySyncSession | None = None

`session_getter`

python

session_getter: SyncSessionGetter = default_session_getter

`type`

python

type: type[T]

`update_by_filter`

python

update_by_filter(data: Input[InputModel], filter_input: BooleanFilterDTO) -> GraphQLResult[InputModel, T]

Asynchronously update entities matching the given filter.

Parameters:

Name	Type	Description	Default
`data`	`Input[InputModel]`	The input data containing fields to update	required
`filter_input`	`BooleanFilterDTO`	The filter criteria to select entities to update	required

Returns:

Type	Description
`GraphQLResult[InputModel, T]`	A GraphQLResult containing the updated entities

`update_by_id`

python

update_by_id(data: Input[InputModel]) -> GraphQLResult[InputModel, T]

Asynchronously update an entity by its ID.

Parameters:

Name	Type	Description	Default
`data`	`Input[InputModel]`	The input data containing the ID and fields to update	required

Returns:

Type	Description
`GraphQLResult[InputModel, T]`	A GraphQLResult containing the updated entity

Raises:

Type	Description
`NoResultFound`	If no entity with the given ID exists

`upsert`

python

upsert(data: Input[InputModel], filter_input: BooleanFilterDTO | None = None, update_fields: builtins.list[EnumDTO] | None = None, conflict_fields: EnumDTO | None = None) -> GraphQLResult[InputModel, T]

Asynchronously insert or update an entity.

Parameters:

Name	Type	Description	Default
`data`	`Input[InputModel]`	The input data for the entity	required
`filter_input`	`BooleanFilterDTO \| None`	Optional filter to find existing entity	`None`
`update_fields`	`list[EnumDTO] \| None`	Optional fields to update if entity exists	`None`
`conflict_fields`	`EnumDTO \| None`	Optional fields to detect conflicts on	`None`

Returns:

Type	Description
`GraphQLResult[InputModel, T]`	A GraphQLResult containing the upserted entity

Repositories ​

StrawchemyAsyncRepository ​

auto_snake_case ​

create ​

default_order_by ​

delete ​

deterministic_ordering ​

execution_options ​

filter_statement ​

get_by_id ​

get_one ​

get_one_or_none ​

graphql_repository ​

info ​

is_async ​

list ​

query_hook ​

root_aggregations ​

session ​

session_getter ​

type ​

update_by_filter ​

update_by_id ​

upsert ​

StrawchemySyncRepository ​

auto_snake_case ​

create ​

default_order_by ​

delete ​

deterministic_ordering ​

execution_options ​

filter_statement ​

get_by_id ​

get_one ​

get_one_or_none ​

graphql_repository ​

info ​

is_async ​

list ​

query_hook ​

root_aggregations ​

session ​

session_getter ​

type ​

update_by_filter ​

update_by_id ​

upsert ​

Repositories

`StrawchemyAsyncRepository`

`auto_snake_case`

`create`

`default_order_by`

`delete`

`deterministic_ordering`

`execution_options`

`filter_statement`

`get_by_id`

`get_one`

`get_one_or_none`

`graphql_repository`

`info`

`is_async`

`list`

`query_hook`

`root_aggregations`

`session`

`session_getter`

`type`

`update_by_filter`

`update_by_id`

`upsert`

`StrawchemySyncRepository`

`auto_snake_case`

`create`

`default_order_by`

`delete`

`deterministic_ordering`

`execution_options`

`filter_statement`

`get_by_id`

`get_one`

`get_one_or_none`

`graphql_repository`

`info`

`is_async`

`list`

`query_hook`

`root_aggregations`

`session`

`session_getter`

`type`

`update_by_filter`

`update_by_id`

`upsert`