Package fastapi_jsonapi index

fastapi_jsonapi.data_layers.fields.enum module

Base enum module.

class fastapi_jsonapi.data_layers.fields.enum.Enum(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: MixinEnum

Base enum class.

All used non-integer enumerations must inherit from this class.

class fastapi_jsonapi.data_layers.fields.enum.IntEnum(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: MixinIntEnum

Base IntEnum class.

All used integer enumerations must inherit from this class.

fastapi_jsonapi.data_layers.fields.mixins module

Enum mixin module.

class fastapi_jsonapi.data_layers.fields.mixins.MixinEnum(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Extension over enum class from standard library.

classmethod inverse()

Return all inverted items sequence.

classmethod keys()

Get all field keys from Enum.

classmethod names()

Get all field names.

classmethod value_to_enum(value)

Convert value to enum.

classmethod values()

Get all values from Enum.

class fastapi_jsonapi.data_layers.fields.mixins.MixinIntEnum(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: IntEnum

Здесь пришлось дублировать код, чтобы обеспечить совместимость с FastAPI и Pydantic.

Основная проблема - данные либы определяют валидаторы для стандартной библиотеки enum, используя вызов issubclass. И для стандартного IntEnum есть отдельная ветка issubclass(IntEnum), в которой происходят специальные преобразования, например, аргументы из запроса конвертируются в тип int. Поэтому OurEnum(int, Enum) не срабатывает по условию issubclass(obj, IntEnum) и выбираются неверные валидаторы и конверторы. А код ниже пришлось задублировать, так как у стандартного Enum есть метакласс, который разрешает только такую цепочку наследования: NewEnum(клас_тип, миксин_без_типа_1, …, миксин_без_типа_n, Enum) По этому правилу нельзя построить наследование, добавляющее миксин без типа к стандартному IntEnum: NewEnum(our_mixin, IntEnum), так как IntEnum = (int, Enum) Поэтому пока остается такое решение до каких-либо исправлений со стороны разработчиков либы, либо появления более гениальных идей

classmethod inverse()

Return all inverted items sequence.

classmethod keys()

Get all field keys from Enum.

classmethod names()

Get all field names.

classmethod value_to_enum(value)

Convert value to enum.

classmethod values()

Get all values from Enum.

fastapi_jsonapi.data_layers.filtering.sqlalchemy module

Helper to create sqlalchemy filters according to filter querystring parameter

class fastapi_jsonapi.data_layers.filtering.sqlalchemy.RelationshipFilteringInfo(*, target_schema: Type[TypeSchema], model: Type[TypeModel], aliased_model: AliasedClass, join_column: InstrumentedAttribute)

Bases: BaseModel

class Config

Bases: object

arbitrary_types_allowed = True
aliased_model: AliasedClass
join_column: InstrumentedAttribute
model: Type[TypeModel]
target_schema: Type[TypeSchema]
fastapi_jsonapi.data_layers.filtering.sqlalchemy.build_filter_expression(schema_field: ModelField, model_column: InstrumentedAttribute, operator: str, value: Any) → BinaryExpression

Builds sqlalchemy filter expression, like YourModel.some_field == value

Custom sqlalchemy filtering logic can be created in a schemas field for any operator To implement a new filtering logic (override existing or create a new one) create a method inside a field following this pattern: _<your_op_name>_sql_filter_

Parameters:

  • schema_field – schemas field instance

  • model_column – sqlalchemy column instance

  • operator – your operator, for example: “eq”, “in”, “ilike_str_array”, …

  • value – filtering value

fastapi_jsonapi.data_layers.filtering.sqlalchemy.build_filter_expressions(filter_item: Dict, target_schema: Type[TypeSchema], target_model: Type[TypeModel], relationships_info: Dict[str, RelationshipFilteringInfo]) → BinaryExpression | BooleanClauseList

Return sqla expressions.

Builds sqlalchemy expression which can be use in where condition: query(Model).where(build_filter_expressions(…))

fastapi_jsonapi.data_layers.filtering.sqlalchemy.build_terminal_node_filter_expressions(filter_item: Dict, target_schema: Type[TypeSchema], target_model: Type[TypeModel], relationships_info: Dict[str, RelationshipFilteringInfo])
fastapi_jsonapi.data_layers.filtering.sqlalchemy.cast_iterable_with_pydantic(types: List[Type], values: List, schema_field: ModelField) → Tuple[List, List[str]]
fastapi_jsonapi.data_layers.filtering.sqlalchemy.cast_value_with_pydantic(types: List[Type], value: Any, schema_field: ModelField) → Tuple[Any | None, List[str]]
fastapi_jsonapi.data_layers.filtering.sqlalchemy.cast_value_with_scheme(field_types: List[Type], value: Any) → Tuple[Any, List[str]]
fastapi_jsonapi.data_layers.filtering.sqlalchemy.check_can_be_none(fields: list[ModelField]) → bool

Return True if None is possible value for target field

fastapi_jsonapi.data_layers.filtering.sqlalchemy.create_filters_and_joins(filter_info: list, model: Type[TypeModel], schema: Type[TypeSchema])
fastapi_jsonapi.data_layers.filtering.sqlalchemy.gather_relationship_paths(filter_item: dict | list) → Set[str]

Extracts relationship paths from query filter

fastapi_jsonapi.data_layers.filtering.sqlalchemy.gather_relationships(entrypoint_model: Type[TypeModel], schema: Type[TypeSchema], relationship_paths: Set[str]) → dict[str, RelationshipFilteringInfo]
fastapi_jsonapi.data_layers.filtering.sqlalchemy.gather_relationships_info(model: Type[TypeModel], schema: Type[TypeSchema], relationship_path: List[str], collected_info: dict[str, RelationshipFilteringInfo], target_relationship_idx: int = 0, prev_aliased_model: Any | None = None) → dict[str, RelationshipFilteringInfo]
fastapi_jsonapi.data_layers.filtering.sqlalchemy.get_custom_filter_expression_callable(schema_field, operator: str) → Callable
fastapi_jsonapi.data_layers.filtering.sqlalchemy.get_model_column(model: Type[TypeModel], schema: Type[TypeSchema], field_name: str) → InstrumentedAttribute
fastapi_jsonapi.data_layers.filtering.sqlalchemy.get_operator(model_column: InstrumentedAttribute, operator_name: str) → str

Get the function operator from his name

Return callable:

a callable to make operation on a column

fastapi_jsonapi.data_layers.filtering.sqlalchemy.is_relationship_filter(name: str) → bool
fastapi_jsonapi.data_layers.filtering.sqlalchemy.is_terminal_node(filter_item: dict) → bool

If node shape is:

{

“name: …, “op: …, “val: …,

}

fastapi_jsonapi.data_layers.filtering.sqlalchemy.prepare_relationships_info(model: Type[TypeModel], schema: Type[TypeSchema], filter_info: list)
fastapi_jsonapi.data_layers.filtering.sqlalchemy.separate_types(types: List[Type]) → Tuple[List[Type], List[Type]]

Separates the types into two kinds.

The first are those for which there are already validators defined by pydantic - str, int, datetime and some other built-in types. The second are all other types for which the arbitrary_types_allowed config is applied when defining the pydantic model

fastapi_jsonapi.data_layers.filtering.sqlalchemy.validator_requires_model_field(validator: Callable) → bool

Check if validator accepts the field param

Parameters:

validator

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation module

Previously used: ‘__’

class fastapi_jsonapi.data_layers.filtering.tortoise_operation.ProcessTypeOperationFieldName(*args, **kwargs)

Bases: Protocol

fastapi_jsonapi.data_layers.filtering.tortoise_operation.add_suffix(field_name: str, suffix: str, sep: str = '__') → str

joins str

Parameters:

  • field_name

  • suffix

  • sep

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.prepare_field_name_for_filtering(field_name: str, type_op: str) → str

Prepare fields for use in ORM.

Parameters:

  • field_name – name of the field by which the filtering will be performed.

  • type_op – operation type.

Returns:

prepared name field.

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_any(field_name: str, type_op: str) → str

used to filter on to many relationships

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_between(field_name: str, type_op: str) → str

used to filter a field between two values

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_contains(field_name: str, type_op: str) → str

field contains specified substring

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_endswith(field_name: str, type_op: str) → str

check if field ends with a string

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_eq(field_name: str, type_op: str) → str

check if field is equal to something

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_ge(field_name: str, type_op: str) → str

check if field is greater than or equal to something

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_gt(field_name: str, type_op: str) → str

check if field is greater than to something

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_has(field_name: str, type_op: str) → str

used to filter on to one relationship

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_icontains(field_name: str, type_op: str) → str

case insensitive contains

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_iendswith(field_name: str, type_op: str) → str

check if field ends with a string

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_iequals(field_name: str, type_op: str) → str

case insensitive equals

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_ilike(field_name: str, type_op: str) → str

case insensitive contains

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_in_(field_name: str, type_op: str) → str

check if field is in a list of values

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_is_(field_name: str, type_op: str) → str

check if field is null. wtf

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_isnot(field_name: str, type_op: str) → str

check if field is not null. wtf

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_istartswith(field_name: str, type_op: str) → str

check if field starts with a string (case insensitive)

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_le(field_name: str, type_op: str) → str

check if field is less than or equal to something

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_like(field_name: str, type_op: str) → str

field contains specified substring

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_lt(field_name: str, type_op: str) → str

check if field is less than to something

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_match(field_name: str, type_op: str) → str

check if field match against a string or pattern

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_ne(field_name: str, type_op: str) → str

check if field is not equal to something

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_notilike(field_name: str, type_op: str) → str

check if field does not contains a string (case insensitive)

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_notin_(field_name: str, type_op: str) → str

check if field is not in a list of values

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_notlike(field_name: str, type_op: str) → str

check if field does not contains a string

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_operation.type_op_startswith(field_name: str, type_op: str) → str

check if field starts with value

Parameters:

  • field_name

  • type_op

Returns:

fastapi_jsonapi.data_layers.filtering.tortoise_orm module

Tortoise filters creator.

class fastapi_jsonapi.data_layers.filtering.tortoise_orm.FilterTortoiseORM(model: TypeModel)

Bases: object

create_query(filter_q: tuple | Q) → Q

Tortoise filter creation.

filter_converter(schema: Type[BaseModel], filters: List[Dict[str, str | int | float | dict | list | None]]) → List

Make a list with filters, which can be used in the tortoise filter.

Parameters:

  • schema – schemas schema of object.

  • filters – list of JSON API filters.

Returns:

list of filters, prepared for use in tortoise model.

Raises:

InvalidFilters – if the filter was created with an error.

async json_api_filter(query, schema: Type[BaseModel], query_params: QueryStringManager) → QuerySet

Make queries with filtering from request.

orm_and_or(op: DBORMOperandType, filters: list) → None | QuerySet | Dict[str, QuerySet | List[QuerySet]]

Filter for query to ORM.

val_to_query(val: Any) → Any

Value to query.

validate(filter_q: None | Q | Dict[str, Q | List[Q]]) → Q | None

Tortoise filter validation.

Parameters:

filter_q – dict with filter body.

Returns:

validated filter.

Raises:

QueryError – if the field in the filter does not match the field in tortoise.

fastapi_jsonapi.data_layers.filtering.tortoise_orm.prepare_filter_pair(field: Type[ModelField], field_name: str, type_op: str, value: Any) → Tuple

Prepare filter.

fastapi_jsonapi.data_layers.sorting.sqlalchemy module

Helper to create sqlalchemy sortings according to filter querystring parameter

class fastapi_jsonapi.data_layers.sorting.sqlalchemy.Node(model: Type[TypeModel], sort_: dict, schema: Type[TypeSchema])

Bases: object

Helper to recursively create sorts with sqlalchemy according to sort querystring parameter

property column: InstrumentedAttribute

Get the column object.

Returns:

the column to filter on

classmethod create_sort(schema_field: ModelField, model_column, order: str)

Create sqlalchemy sort.

Params schema_field:

Params model_column:

column sqlalchemy

Params order:

desc | asc (or custom)

Returns:

property name: str

Return the name of the node or raise a BadRequest exception

Return str:

the name of the sort to sort on

property related_model: DeclarativeMeta

Get the related model of a relationship field.

Returns:

the related model.

property related_schema: Type[TypeSchema]

Get the related schema of a relationship field.

Returns:

the related schema

resolve() → Tuple[BinaryExpression, List[List[Any]]]

Create sort for a particular node of the sort tree.

fastapi_jsonapi.data_layers.sorting.sqlalchemy.create_sorts(model: Type[TypeModel], filter_info: list | dict, schema: Type[TypeSchema])

Apply filters from filters information to base query.

Params model:

the model of the node.

Params filter_info:

current node filter information.

Params schema:

the resource.

fastapi_jsonapi.data_layers.sorting.tortoise_orm module

class fastapi_jsonapi.data_layers.sorting.tortoise_orm.SortTortoiseORM

Bases: object

classmethod sort(query: QuerySet, query_params_sorting: List[Dict[str, str]], default_sort: str = '') → QuerySet

Реализация динамической сортировки для query.

Parameters:

  • query – запрос

  • query_params_sorting – параметры от клиента

  • default_sort – дефолтная сортировка, например “-id” или sort=-id,created_at

fastapi_jsonapi.data_layers.base module

The base class of a data layer.

If you want to create your own data layer you must inherit from this base class

class fastapi_jsonapi.data_layers.base.BaseDataLayer(request: Request, schema: Type[TypeSchema], model: Type[TypeModel], url_id_field: str, id_name_field: str | None = None, disable_collection_count: bool = False, default_collection_count: int = -1, type_: str = '', **kwargs)

Bases: object

Base class of a data layer

async after_create_object(obj, data, view_kwargs)

Provide additional data after object creation

Parameters:

  • obj – an object from data layer

  • data – the data validated by schemas

  • view_kwargs – kwargs from the resource view

async after_create_relationship(obj, updated, json_data, relationship_field, related_id_field, view_kwargs)

Make work after to create a relationship

Parameters:

  • obj – an object from data layer

  • updated (bool) – True if object was updated else False

  • json_data – the request params

  • relationship_field (str) – the model attribute used for relationship

  • related_id_field (str) – the identifier field of the related model

  • view_kwargs – kwargs from the resource view

Return boolean:

True if relationship have changed else False

async after_delete_object(obj: TypeModel, view_kwargs)

Make work after delete object

Parameters:

  • obj – an object from data layer

  • view_kwargs – kwargs from the resource view

async after_delete_objects(objects: List[TypeModel], view_kwargs: dict)

Any action after deleting objects.

Parameters:

  • objects – an object from data layer.

  • view_kwargs – kwargs from the resource view.

async after_delete_relationship(obj, updated, json_data, relationship_field, related_id_field, view_kwargs)

Make work after to delete a relationship

Parameters:

  • obj – an object from data layer

  • updated (bool) – True if object was updated else False

  • json_data – the request params

  • relationship_field (str) – the model attribute used for relationship

  • related_id_field (str) – the identifier field of the related model

  • view_kwargs – kwargs from the resource view

async after_get_collection(collection, qs, view_kwargs)

Make work after to retrieve a collection of objects

Parameters:

  • collection (iterable) – the collection of objects

  • qs – a querystring manager to retrieve information from url

  • view_kwargs – kwargs from the resource view

async after_get_object(obj, view_kwargs)

Make work after to retrieve an object

Parameters:

  • obj – an object from data layer

  • view_kwargs – kwargs from the resource view

async after_get_relationship(obj, related_objects, relationship_field, related_type_, related_id_field, view_kwargs)

Make work after to get information about a relationship

Parameters:

  • obj – an object from data layer

  • related_objects (iterable) – related objects of the object

  • relationship_field (str) – the model attribute used for relationship

  • related_type (str) – the related resource type

  • related_id_field (str) – the identifier field of the related model

  • view_kwargs – kwargs from the resource view

Return tuple:

the object and related object(s)

async after_update_object(obj: TypeModel, data, view_kwargs)

Make work after update object

Parameters:

  • obj – an object from data layer

  • data – the data validated by schemas

  • view_kwargs – kwargs from the resource view

async after_update_relationship(obj, updated, json_data, relationship_field, related_id_field, view_kwargs)

Make work after to update a relationship

Parameters:

  • obj – an object from data layer

  • updated (bool) – True if object was updated else False

  • json_data – the request params

  • relationship_field (str) – the model attribute used for relationship

  • related_id_field (str) – the identifier field of the related model

  • view_kwargs – kwargs from the resource view

Return boolean:

True if relationship have changed else False

async atomic_end(success: bool = True)
async atomic_start(previous_dl: BaseDataLayer | None = None)
async before_create_object(data, view_kwargs)

Provide additional data before object creation

Parameters:

  • data – the data validated by schemas

  • view_kwargs – kwargs from the resource view

async before_create_relationship(json_data, relationship_field, related_id_field, view_kwargs)

Make work before to create a relationship

Parameters:

  • json_data – the request params

  • relationship_field (str) – the model attribute used for relationship

  • related_id_field (str) – the identifier field of the related model

  • view_kwargs – kwargs from the resource view

Return boolean:

True if relationship have changed else False

async before_delete_object(obj: TypeModel, view_kwargs)

Make checks before delete object

Parameters:

  • obj – an object from data layer

  • view_kwargs – kwargs from the resource view

async before_delete_objects(objects: List[TypeModel], view_kwargs: dict)

Make checks before deleting objects.

Parameters:

  • objects – an object from data layer.

  • view_kwargs – kwargs from the resource view.

async before_delete_relationship(json_data, relationship_field, related_id_field, view_kwargs)

Make work before to delete a relationship

Parameters:

  • json_data – the request params

  • relationship_field (str) – the model attribute used for relationship

  • related_id_field (str) – the identifier field of the related model

  • view_kwargs – kwargs from the resource view

async before_get_collection(qs, view_kwargs)

Make work before to retrieve a collection of objects

Parameters:

  • qs – a querystring manager to retrieve information from url

  • view_kwargs – kwargs from the resource view

async before_get_object(view_kwargs)

Make work before to retrieve an object

Parameters:

view_kwargs – kwargs from the resource view

async before_get_relationship(relationship_field, related_type_, related_id_field, view_kwargs)

Make work before to get information about a relationship

Parameters:

  • relationship_field (str) – the model attribute used for relationship

  • related_type (str) – the related resource type

  • related_id_field (str) – the identifier field of the related model

  • view_kwargs – kwargs from the resource view

Return tuple:

the object and related object(s)

async before_update_object(obj, data, view_kwargs)

Make checks or provide additional data before update object

Parameters:

  • obj – an object from data layer

  • data – the data validated by schemas

  • view_kwargs – kwargs from the resource view

async before_update_relationship(json_data, relationship_field, related_id_field, view_kwargs)

Make work before to update a relationship

Parameters:

  • json_data – the request params

  • relationship_field (str) – the model attribute used for relationship

  • related_id_field (str) – the identifier field of the related model

  • view_kwargs – kwargs from the resource view

Return boolean:

True if relationship have changed else False

async create_object(data_create: BaseJSONAPIItemInSchema, view_kwargs: dict) → TypeModel

Create an object

Parameters:

  • data_create – validated data

  • view_kwargs – kwargs from the resource view

Return DeclarativeMeta:

an object

async create_relationship(json_data, relationship_field, related_id_field, view_kwargs)

Create a relationship

Parameters:

  • json_data – the request params

  • relationship_field (str) – the model attribute used for relationship

  • related_id_field (str) – the identifier field of the related model

  • view_kwargs – kwargs from the resource view

Return boolean:

True if relationship have changed else False

async delete_object(obj, view_kwargs)

Delete an item through the data layer

Parameters:

  • obj (DeclarativeMeta) – an object

  • view_kwargs – kwargs from the resource view

async delete_objects(objects: List[TypeModel], view_kwargs)
async delete_relationship(json_data, relationship_field, related_id_field, view_kwargs)

Delete a relationship

Parameters:

  • json_data – the request params

  • relationship_field (str) – the model attribute used for relationship

  • related_id_field (str) – the identifier field of the related model

  • view_kwargs – kwargs from the resource view

async get_collection(qs: QueryStringManager, view_kwargs: dict | None = None) → Tuple[int, list]

Retrieve a collection of objects

Parameters:

  • qs – a querystring manager to retrieve information from url

  • view_kwargs – kwargs from the resource view

Return tuple:

the number of object and the list of objects

async get_object(view_kwargs: dict, qs: QueryStringManager | None = None) → TypeModel

Retrieve an object

Parameters:

  • view_kwargs – kwargs from the resource view

  • qs

Return DeclarativeMeta:

an object

get_object_id(obj: TypeModel)
get_object_id_field()
get_object_id_field_name()

compound key may cause errors

Returns:

Prepare query for the related model

Parameters:

related_model – Related ORM model class (not instance)

Returns:

Get related object.

Parameters:

  • related_model – Related ORM model class (not instance)

  • related_id_field – id field of the related model (usually it’s id)

  • id_value – related object id value

Returns:

an ORM object

Prepare query to get related object

Parameters:

  • related_model

  • related_id_field

  • id_value

Returns:

Get related objects list.

Parameters:

  • related_model – Related ORM model class (not instance)

  • related_id_field – id field of the related model (usually it’s id)

  • ids – related object id values list

Returns:

a list of ORM objects

Prepare query to get related objects list

Parameters:

  • related_model

  • related_id_field

  • ids

Returns:

async get_relationship(relationship_field, related_type_, related_id_field, view_kwargs)

Get information about a relationship

Parameters:

  • relationship_field (str) – the model attribute used for relationship

  • related_type (str) – the related resource type

  • related_id_field (str) – the identifier field of the related model

  • view_kwargs – kwargs from the resource view

Return tuple:

the object and related object(s)

query(view_kwargs)

Construct the base query to retrieve wanted data

Parameters:

view_kwargs – kwargs from the resource view

async update_object(obj, data_update: BaseJSONAPIItemInSchema, view_kwargs: dict)

Update an object

Parameters:

  • obj – an object

  • data_update – the data validated by schemas

  • view_kwargs – kwargs from the resource view

Return boolean:

True if object have changed else False

async update_relationship(json_data, relationship_field, related_id_field, view_kwargs)

Update a relationship

Parameters:

  • json_data – the request params

  • relationship_field (str) – the model attribute used for relationship

  • related_id_field (str) – the identifier field of the related model

  • view_kwargs – kwargs from the resource view

Return boolean:

True if relationship have changed else False

fastapi_jsonapi.data_typing module

fastapi_jsonapi.data_layers.orm module

ORM types enums.

class fastapi_jsonapi.data_layers.orm.DBORMOperandType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

and_ = 'and'
not_ = 'not'
or_ = 'or'

fastapi_jsonapi.data_layers.shared module

fastapi_jsonapi.data_layers.shared.create_filters_or_sorts(model: Type[TypeModel], filter_or_sort_info: list | dict, class_node: Type[NodeSQLAlchemy], schema: Type[TypeSchema]) → Tuple

Apply filters / sorts from filters / sorts information to base query

Parameters:

  • model – the model of the node

  • filter_or_sort_info – current node filter_or_sort information

  • class_node

  • schema – the resource

fastapi_jsonapi.data_layers.sqla_orm module

This module is a CRUD interface between resource managers and the sqlalchemy ORM

class fastapi_jsonapi.data_layers.sqla_orm.SqlalchemyDataLayer(schema: Type[TypeSchema], model: Type[TypeModel], session: AsyncSession, disable_collection_count: bool = False, default_collection_count: int = -1, id_name_field: str | None = None, url_id_field: str = 'id', eagerload_includes: bool = True, query: Select | None = None, auto_convert_id_to_column_type: bool = True, **kwargs: Any)

Bases: BaseDataLayer

Sqlalchemy data layer

async after_create_object(obj: TypeModel, model_kwargs: dict, view_kwargs: dict)

Provide additional data after object creation.

Parameters:

  • obj – an object from data layer.

  • model_kwargs – the data validated by pydantic.

  • view_kwargs – kwargs from the resource view.

async after_create_relationship(obj: Any, updated: bool, json_data: dict, relationship_field: str, related_id_field: str, view_kwargs: dict)

Make work after to create a relationship.

Parameters:

  • obj – an object from data layer.

  • updated – True if object was updated else False.

  • json_data – the request params.

  • relationship_field – the model attribute used for relationship.

  • related_id_field – the identifier field of the related model.

  • view_kwargs – kwargs from the resource view.

Return boolean:

True if relationship have changed else False.

async after_delete_object(obj: TypeModel, view_kwargs: dict)

Make work after delete object.

Parameters:

  • obj – an object from data layer.

  • view_kwargs – kwargs from the resource view.

async after_delete_objects(objects: List[TypeModel], view_kwargs: dict)

Any actions after deleting objects.

Parameters:

  • objects – an object from data layer.

  • view_kwargs – kwargs from the resource view.

async after_delete_relationship(obj: Any, updated: bool, json_data: dict, relationship_field: str, related_id_field: str, view_kwargs: dict)

Make work after to delete a relationship.

Parameters:

  • obj – an object from data layer.

  • updated – True if object was updated else False.

  • json_data – the request params.

  • relationship_field – the model attribute used for relationship.

  • related_id_field – the identifier field of the related model.

  • view_kwargs – kwargs from the resource view.

async after_get_collection(collection: Iterable, qs: QueryStringManager, view_kwargs: dict)

Make work after to retrieve a collection of objects.

Parameters:

  • collection – the collection of objects.

  • qs – a querystring manager to retrieve information from url.

  • view_kwargs – kwargs from the resource view.

async after_get_object(obj: Any, view_kwargs: dict)

Make work after to retrieve an object.

Parameters:

  • obj – an object from data layer.

  • view_kwargs – kwargs from the resource view.

async after_get_relationship(obj: Any, related_objects: Iterable, relationship_field: str, related_type_: str, related_id_field: str, view_kwargs: dict)

Make work after to get information about a relationship.

Parameters:

  • obj – an object from data layer.

  • related_objects – related objects of the object.

  • relationship_field – the model attribute used for relationship.

  • related_type – the related resource type.

  • related_id_field – the identifier field of the related model.

  • view_kwargs – kwargs from the resource view.

Return tuple:

the object and related object(s).

async after_update_object(obj: Any, model_kwargs: dict, view_kwargs: dict)

Make work after update object.

Parameters:

  • obj – an object from data layer.

  • model_kwargs – the data validated by schemas.

  • view_kwargs – kwargs from the resource view.

async after_update_relationship(obj: Any, updated: bool, json_data: dict, relationship_field: str, related_id_field: str, view_kwargs: dict)

Make work after to update a relationship.

Parameters:

  • obj – an object from data layer.

  • updated – True if object was updated else False.

  • json_data – the request params.

  • relationship_field – the model attribute used for relationship.

  • related_id_field – the identifier field of the related model.

  • view_kwargs – kwargs from the resource view.

Return boolean:

True if relationship have changed else False.

async apply_relationships(obj: TypeModel, data_create: BaseJSONAPIItemInSchema, action_trigger: Literal['create', 'update']) → None

Handles relationships passed in request

Parameters:

  • obj

  • data_create

  • action_trigger – indicates which one operation triggered relationships applying

Returns:

async atomic_end(success: bool = True)
async atomic_start(previous_dl: SqlalchemyDataLayer | None = None)
async before_create_object(model_kwargs: dict, view_kwargs: dict)

Provide additional data before object creation.

Parameters:

  • model_kwargs – the data validated by pydantic.

  • view_kwargs – kwargs from the resource view.

async before_create_relationship(json_data: dict, relationship_field: str, related_id_field: str, view_kwargs: dict)

Make work before to create a relationship.

Parameters:

  • json_data – the request params.

  • relationship_field – the model attribute used for relationship.

  • related_id_field – the identifier field of the related model.

  • view_kwargs – kwargs from the resource view.

Return boolean:

True if relationship have changed else False.

async before_delete_object(obj: TypeModel, view_kwargs: dict)

Make checks before delete object.

Parameters:

  • obj – an object from data layer.

  • view_kwargs – kwargs from the resource view.

async before_delete_objects(objects: List[TypeModel], view_kwargs: dict)

Make checks before deleting objects.

Parameters:

  • objects – an object from data layer.

  • view_kwargs – kwargs from the resource view.

async before_delete_relationship(json_data: dict, relationship_field: str, related_id_field: str, view_kwargs: dict)

Make work before to delete a relationship.

Parameters:

  • json_data – the request params.

  • relationship_field – the model attribute used for relationship.

  • related_id_field – the identifier field of the related model.

  • view_kwargs – kwargs from the resource view.

async before_get_collection(qs: QueryStringManager, view_kwargs: dict)

Make work before to retrieve a collection of objects.

Parameters:

  • qs – a querystring manager to retrieve information from url.

  • view_kwargs – kwargs from the resource view.

async before_get_object(view_kwargs: dict)

Make work before to retrieve an object.

Parameters:

view_kwargs – kwargs from the resource view.

async before_get_relationship(relationship_field: str, related_type_: str, related_id_field: str, view_kwargs: dict)

Make work before to get information about a relationship.

Parameters:

  • relationship_field (str) – the model attribute used for relationship.

  • related_type (str) – the related resource type.

  • related_id_field (str) – the identifier field of the related model.

  • view_kwargs (dict) – kwargs from the resource view.

Return tuple:

the object and related object(s).

async before_update_object(obj: Any, model_kwargs: dict, view_kwargs: dict)

Make checks or provide additional data before update object.

Parameters:

  • obj – an object from data layer.

  • model_kwargs – the data validated by schemas.

  • view_kwargs – kwargs from the resource view.

async before_update_relationship(json_data: dict, relationship_field: str, related_id_field: str, view_kwargs: dict)

Make work before to update a relationship.

Parameters:

  • json_data – the request params.

  • relationship_field – the model attribute used for relationship.

  • related_id_field – the identifier field of the related model.

  • view_kwargs – kwargs from the resource view.

Return boolean:

True if relationship have changed else False.

async check_object_has_relationship_or_raise(obj: TypeModel, relation_name: str)

Checks that there is relationship with relation_name in obj

Parameters:

  • obj

  • relation_name

async create_object(data_create: BaseJSONAPIItemInSchema, view_kwargs: dict) → TypeModel

Create an object through sqlalchemy.

Parameters:

  • data_create – the data validated by pydantic.

  • view_kwargs – kwargs from the resource view.

Returns:

async create_relationship(json_data: dict, relationship_field: str, related_id_field: str, view_kwargs: dict) → bool

Create a relationship.

Parameters:

  • json_data – the request params.

  • relationship_field – the model attribute used for relationship.

  • related_id_field – the identifier field of the related model.

  • view_kwargs – kwargs from the resource view.

Returns:

True if relationship have changed else False.

async delete_object(obj: TypeModel, view_kwargs: dict)

Delete an object through sqlalchemy.

Parameters:

  • obj – an item from sqlalchemy.

  • view_kwargs – kwargs from the resource view.

async delete_objects(objects: List[TypeModel], view_kwargs: dict)
async delete_relationship(json_data: dict, relationship_field: str, related_id_field: str, view_kwargs: dict)

Delete a relationship.

Parameters:

  • json_data – the request params.

  • relationship_field – the model attribute used for relationship.

  • related_id_field – the identifier field of the related model.

  • view_kwargs – kwargs from the resource view.

eagerload_includes(query: Select, qs: QueryStringManager) → Select

Use eagerload feature of sqlalchemy to optimize data retrieval for include querystring parameter.

Parameters:

  • query – sqlalchemy queryset.

  • qs – a querystring manager to retrieve information from url.

Returns:

the query with includes eagerloaded.

filter_query(query: Select, filter_info: list | None) → Select

Filter query according to jsonapi 1.0.

Parameters:

  • query – sqlalchemy query to sort.

  • filter_info – filter information.

Returns:

the sorted query.

async get_collection(qs: QueryStringManager, view_kwargs: dict | None = None) → Tuple[int, list]

Retrieve a collection of objects through sqlalchemy.

Parameters:

  • qs – a querystring manager to retrieve information from url.

  • view_kwargs – kwargs from the resource view.

Returns:

the number of object and the list of objects.

async get_collection_count(query: Select, qs: QueryStringManager, view_kwargs: dict) → int

Returns number of elements for this collection

Parameters:

  • query – SQLAlchemy query

  • qs – QueryString

  • view_kwargs – view kwargs

Returns:

async get_object(view_kwargs: dict, qs: QueryStringManager | None = None) → TypeModel

Retrieve an object through sqlalchemy.

Parameters:

  • view_kwargs – kwargs from the resource view

  • qs

Return DeclarativeMeta:

an object from sqlalchemy

get_object_id_field_name()

compound key may cause errors

Returns:

Retrieves object or objects to link from database

Parameters:

  • related_model

  • relationship_info

  • relationship_in

Prepare sql query (statement) to fetch related model

Parameters:

related_model

Returns:

Get related object.

Parameters:

  • related_model – SQLA ORM model class

  • related_id_field – id field of the related model (usually it’s id)

  • id_value – related object id value

Returns:

a related SQLA ORM object

Prepare query to get related object

Parameters:

  • related_model

  • related_id_field

  • id_value

Returns:

Fetch related objects (many)

Parameters:

  • related_model

  • related_id_field

  • ids

Returns:

Prepare query to get related objects list

Parameters:

  • related_model

  • related_id_field

  • ids

Returns:

async get_relationship(relationship_field: str, related_type_: str, related_id_field: str, view_kwargs: dict) → Tuple[Any, Any]

Get a relationship.

Parameters:

  • relationship_field – the model attribute used for relationship.

  • related_type – the related resource type.

  • related_id_field – the identifier field of the related model.

  • view_kwargs – kwargs from the resource view.

Returns:

the object and related object(s).

Links target object with relationship object or objects

Parameters:

  • obj

  • relation_name

  • related_data

  • action_trigger – indicates which one operation triggered relationships applying

paginate_query(query: Select, paginate_info: PaginationQueryStringManager) → Select

Paginate query according to jsonapi 1.0.

Parameters:

  • query – sqlalchemy queryset.

  • paginate_info – pagination information.

Returns:

the paginated query

prepare_id_value(col: InstrumentedAttribute, value: Any) → Any

Convert value to the required python type.

Type is declared on the SQLA column.

Parameters:

  • col

  • value

Returns:

query(view_kwargs: dict) → Select

Construct the base query to retrieve wanted data.

Parameters:

view_kwargs – kwargs from the resource view

retrieve_object_query(view_kwargs: dict, filter_field: InstrumentedAttribute, filter_value: Any) → Select

Build query to retrieve object.

Parameters:

  • view_kwargs – kwargs from the resource view

  • filter_field – the field to filter on

  • filter_value – the value to filter with

Return sqlalchemy query:

a query from sqlalchemy

async save()
sort_query(query: Select, sort_info: list) → Select

Sort query according to jsonapi 1.0.

Parameters:

  • query – sqlalchemy query to sort.

  • sort_info – sort information.

Returns:

the sorted query.

async update_object(obj: TypeModel, data_update: BaseJSONAPIItemInSchema, view_kwargs: dict) → bool

Update an object through sqlalchemy.

Parameters:

  • obj – an object from sqlalchemy.

  • data_update – the data validated by pydantic.

  • view_kwargs – kwargs from the resource view.

Returns:

True if object have changed else False.

async update_relationship(json_data: dict, relationship_field: str, related_id_field: str, view_kwargs: dict) → bool

Update a relationship

Parameters:

  • json_data – the request params.

  • relationship_field – the model attribute used for relationship.

  • related_id_field – the identifier field of the related model.

  • view_kwargs – kwargs from the resource view.

Returns:

True if relationship have changed else False.

fastapi_jsonapi.data_layers.tortoise_orm module

This module is a CRUD interface between resource managers and the Tortoise ORM

class fastapi_jsonapi.data_layers.tortoise_orm.TortoiseDataLayer(schema: Type[TypeSchema], model: Type[TypeModel], disable_collection_count: bool = False, default_collection_count: int = -1, id_name_field: str | None = None, url_id_field: str = 'id', query: QuerySet | None = None, **kwargs: Any)

Bases: BaseDataLayer

Tortoise data layer

async after_create_object(obj: Any, data: dict, view_kwargs: dict)

Provide additional data after object creation.

Parameters:

  • obj – an object from data layer.

  • data – the data validated by pydantic.

  • view_kwargs – kwargs from the resource view.

async after_create_relationship(obj: Any, updated: bool, json_data: dict, relationship_field: str, related_id_field: str, view_kwargs: dict)

Make work after to create a relationship.

Parameters:

  • obj – an object from data layer.

  • updated – True if object was updated else False.

  • json_data – the request params.

  • relationship_field – the model attribute used for relationship.

  • related_id_field – the identifier field of the related model.

  • view_kwargs – kwargs from the resource view.

Return boolean:

True if relationship have changed else False.

async after_delete_object(obj: Any, view_kwargs: dict)

Make work after delete object.

Parameters:

  • obj – an object from data layer.

  • view_kwargs – kwargs from the resource view.

async after_delete_relationship(obj: Any, updated: bool, json_data: dict, relationship_field: str, related_id_field: str, view_kwargs: dict)

Make work after to delete a relationship.

Parameters:

  • obj – an object from data layer.

  • updated – True if object was updated else False.

  • json_data – the request params.

  • relationship_field – the model attribute used for relationship.

  • related_id_field – the identifier field of the related model.

  • view_kwargs – kwargs from the resource view.

async after_get_collection(collection: Iterable, qs: QueryStringManager, view_kwargs: dict) → Iterable

Make work after to retrieve a collection of objects.

Parameters:

  • collection – the collection of objects.

  • qs – a querystring manager to retrieve information from url.

  • view_kwargs – kwargs from the resource view.

async after_get_object(obj: Any, view_kwargs: dict)

Make work after to retrieve an object.

Parameters:

  • obj – an object from data layer.

  • view_kwargs – kwargs from the resource view.

async after_get_relationship(obj: Any, related_objects: Iterable, relationship_field: str, related_type_: str, related_id_field: str, view_kwargs: dict)

Make work after to get information about a relationship.

Parameters:

  • obj – an object from data layer.

  • related_objects – related objects of the object.

  • relationship_field – the model attribute used for relationship.

  • related_type – the related resource type.

  • related_id_field – the identifier field of the related model.

  • view_kwargs – kwargs from the resource view.

Return tuple:

the object and related object(s).

async after_update_object(obj: Any, data: dict, view_kwargs: dict)

Make work after update object.

Parameters:

  • obj – an object from data layer.

  • data – the data validated by schemas.

  • view_kwargs – kwargs from the resource view.

async after_update_relationship(obj: Any, updated: bool, json_data: dict, relationship_field: str, related_id_field: str, view_kwargs: dict)

Make work after to update a relationship.

Parameters:

  • obj – an object from data layer.

  • updated – True if object was updated else False.

  • json_data – the request params.

  • relationship_field – the model attribute used for relationship.

  • related_id_field – the identifier field of the related model.

  • view_kwargs – kwargs from the resource view.

Return boolean:

True if relationship have changed else False.

async before_create_object(data: dict, view_kwargs: dict)

Provide additional data before object creation.

Parameters:

  • data – the data validated by pydantic.

  • view_kwargs – kwargs from the resource view.

async before_create_relationship(json_data: dict, relationship_field: str, related_id_field: str, view_kwargs: dict)

Make work before to create a relationship.

Parameters:

  • json_data – the request params.

  • relationship_field – the model attribute used for relationship.

  • related_id_field – the identifier field of the related model.

  • view_kwargs – kwargs from the resource view.

Return boolean:

True if relationship have changed else False.

async before_delete_object(obj: Any, view_kwargs: dict)

Make checks before delete object.

Parameters:

  • obj – an object from data layer.

  • view_kwargs – kwargs from the resource view.

async before_delete_relationship(json_data: dict, relationship_field: str, related_id_field: str, view_kwargs: dict)

Make work before to delete a relationship.

Parameters:

  • json_data – the request params.

  • relationship_field – the model attribute used for relationship.

  • related_id_field – the identifier field of the related model.

  • view_kwargs – kwargs from the resource view.

async before_get_collection(qs: QueryStringManager, view_kwargs: dict)

Make work before to retrieve a collection of objects.

Parameters:

  • qs – a querystring manager to retrieve information from url.

  • view_kwargs – kwargs from the resource view.

async before_get_object(view_kwargs: dict)

Make work before to retrieve an object.

Parameters:

view_kwargs – kwargs from the resource view.

async before_get_relationship(relationship_field: str, related_type_: str, related_id_field: str, view_kwargs: dict)

Make work before to get information about a relationship.

Parameters:

  • relationship_field (str) – the model attribute used for relationship.

  • related_type (str) – the related resource type.

  • related_id_field (str) – the identifier field of the related model.

  • view_kwargs (dict) – kwargs from the resource view.

Return tuple:

the object and related object(s).

async before_update_object(obj: Any, data: dict, view_kwargs: dict)

Make checks or provide additional data before update object.

Parameters:

  • obj – an object from data layer.

  • data – the data validated by schemas.

  • view_kwargs – kwargs from the resource view.

async before_update_relationship(json_data: dict, relationship_field: str, related_id_field: str, view_kwargs: dict)

Make work before to update a relationship.

Parameters:

  • json_data – the request params.

  • relationship_field – the model attribute used for relationship.

  • related_id_field – the identifier field of the related model.

  • view_kwargs – kwargs from the resource view.

Return boolean:

True if relationship have changed else False.

async create_object(data_create: BaseJSONAPIItemInSchema, view_kwargs: dict) → TypeModel

Create an object

Parameters:

  • data_create – validated data

  • view_kwargs – kwargs from the resource view

Return DeclarativeMeta:

an object

async create_relationship(json_data: dict, relationship_field: str, related_id_field: str, view_kwargs: dict) → bool

Create a relationship.

Parameters:

  • json_data – the request params.

  • relationship_field – the model attribute used for relationship.

  • related_id_field – the identifier field of the related model.

  • view_kwargs – kwargs from the resource view.

Returns:

True if relationship have changed else False.

async delete_object(obj: TypeModel, view_kwargs: dict)

Delete an object through Tortoise.

Parameters:

  • obj – an item from Tortoise.

  • view_kwargs – kwargs from the resource view.

async delete_relationship(json_data: dict, relationship_field: str, related_id_field: str, view_kwargs: dict)

Delete a relationship.

Parameters:

  • json_data – the request params.

  • relationship_field – the model attribute used for relationship.

  • related_id_field – the identifier field of the related model.

  • view_kwargs – kwargs from the resource view.

eagerload_includes(query: QuerySet, qs: QueryStringManager) → QuerySet

Use eagerload feature of Tortoise to optimize data retrieval for include querystring parameter.

Parameters:

  • query – Tortoise queryset.

  • qs – a querystring manager to retrieve information from url.

Returns:

the query with includes eagerloaded.

async get_collection(qs: QueryStringManager, view_kwargs: dict | None = None) → Tuple[int, list]

Retrieve a collection of objects through Tortoise.

Parameters:

  • qs – a querystring manager to retrieve information from url.

  • view_kwargs – kwargs from the resource view.

Returns:

the number of object and the list of objects.

async get_collection_count(query: QuerySet) → int

Prepare query to fetch collection

Parameters:

  • query – Tortoise query

  • qs – QueryString

  • view_kwargs – view kwargs

Returns:

async get_object(view_kwargs: dict, qs: QueryStringManager | None = None) → TypeModel

Retrieve an object

Parameters:

  • view_kwargs – kwargs from the resource view

  • qs

Return DeclarativeMeta:

an object

Get related object.

Parameters:

  • related_model – Tortoise model

  • related_id_field – the identifier field of the related model

  • id_value – related object id value

Returns:

a related object

async get_relationship(relationship_field: str, related_type_: str, related_id_field: str, view_kwargs: dict) → Tuple[Any, Any]

Get a relationship.

Parameters:

  • relationship_field – the model attribute used for relationship.

  • related_type – the related resource type.

  • related_id_field – the identifier field of the related model.

  • view_kwargs – kwargs from the resource view.

Returns:

the object and related object(s).

paginate_query(query: QuerySet, paginate_info: PaginationQueryStringManager) → QuerySet

Paginate query according to jsonapi 1.0.

Parameters:

  • query – Tortoise queryset.

  • paginate_info – pagination information.

Returns:

the paginated query

query(view_kwargs: dict) → QuerySet

Construct the base query to retrieve wanted data.

Parameters:

view_kwargs – kwargs from the resource view

retrieve_object_query(view_kwargs: dict, filter_field: Any, filter_value: Any) → QuerySet

Build query to retrieve object.

Parameters:

  • view_kwargs – kwargs from the resource view

  • filter_field – the field to filter on

  • filter_value – the value to filter with

Return Tortoise query:

a query from Tortoise

async update_object(obj: TypeModel, data_update: BaseJSONAPIItemInSchema, view_kwargs: dict) → bool

Update an object through Tortoise.

Parameters:

  • obj – an object from Tortoise.

  • data – the data validated by schemas.

  • view_kwargs – kwargs from the resource view.

Returns:

True if object have changed else False.

async update_relationship(json_data: dict, relationship_field: str, related_id_field: str, view_kwargs: dict) → bool

Update a relationship

Parameters:

  • json_data – the request params.

  • relationship_field – the model attribute used for relationship.

  • related_id_field – the identifier field of the related model.

  • view_kwargs – kwargs from the resource view.

Returns:

True if relationship have changed else False.

fastapi_jsonapi.api module

JSON API router class.

class fastapi_jsonapi.api.RoutersJSONAPI(router: APIRouter, path: str | List[str], tags: List[str], class_list: Type[ListViewBase], class_detail: Type[DetailViewBase], model: Type[TypeModel], schema: Type[BaseModel], resource_type: str, schema_in_post: Type[BaseModel] | None = None, schema_in_patch: Type[BaseModel] | None = None, pagination_default_size: int | None = 25, pagination_default_number: int | None = 1, pagination_default_offset: int | None = None, pagination_default_limit: int | None = None, methods: Iterable[str] = (), max_cache_size: int = 0)

Bases: object

API Router interface for JSON API endpoints in web-services.

DEFAULT_METHODS = ('ViewMethods.GET_LIST', 'ViewMethods.POST', 'ViewMethods.DELETE_LIST', 'ViewMethods.GET', 'ViewMethods.DELETE', 'ViewMethods.PATCH')
Methods

alias of ViewMethods

all_jsonapi_routers: ClassVar[Dict[str, RoutersJSONAPI]] = {}
get_endpoint_name(action: Literal['get', 'create', 'update', 'delete'], kind: Literal['list', 'detail'])

Generate view name

:param action :param kind: list / detail :return:

async handle_view_dependencies(request: Request, view_cls: Type[ViewBase], method: HTTPMethod) → Dict[str, Any]

Combines all dependencies (prepared) and returns them as list

Consider method config is already prepared for generic views Reuse the same config for atomic operations

Parameters:

  • request

  • view_cls

  • method

Returns:

prepare_dependencies_handler_signature(custom_handler: Callable[[...], Any], method_config: HTTPMethodConfig) → Signature
class fastapi_jsonapi.api.ViewMethods(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

DELETE = '5'
DELETE_LIST = '3'
GET = '4'
GET_LIST = '1'
PATCH = '6'
POST = '2'

fastapi_jsonapi.jsonapi_typing module

JSON API types.

fastapi_jsonapi.querystring module

Helper to deal with querystring parameters according to jsonapi specification.

class fastapi_jsonapi.querystring.HeadersQueryStringManager(*, host: str | None = None, connection: str | None = None, accept: str | None = None, referer: str | None = None, **extra_data: Any)

Bases: BaseModel

Header query string manager.

Contains info about request headers.

accept: str | None
accept_encoding: str | None
accept_language: str | None
connection: str | None
host: str | None
referer: str | None
user_agent: str | None
class fastapi_jsonapi.querystring.PaginationQueryStringManager(*, offset: int | None = None, size: int | None = 25, number: int = 1, limit: int | None = None)

Bases: BaseModel

Pagination query string manager.

Contains info about offsets, sizes, number and limits of query with pagination.

limit: int | None
number: int
offset: int | None
size: int | None
class fastapi_jsonapi.querystring.QueryStringManager(request: Request)

Bases: object

Querystring parser according to jsonapi reference.

property fields: Dict[str, List[str]]

Return fields wanted by client.

Returns:

a dict of sparse fieldsets information

Return value will be a dict containing all fields by resource, for example:

{
    "user": ['name', 'email'],
}
Raises:

InvalidField – if result field not in schema.

property filters: List[dict]

Return filters from query string.

Returns:

filter information

Raises:

InvalidFilters – if filter loading from json has failed.

get_sorts(schema: Type[TypeSchema]) → List[Dict[str, str]]

Return fields to sort by including sort name for SQLAlchemy and row sort parameter for other ORMs.

Returns:

a list of sorting information

Example of return value:

[
    {'field': 'created_at', 'order': 'desc'},
]
Raises:

InvalidSort – if sort field wrong.

property include: List[str]

Return fields to include.

Returns:

a list of include information.

Raises:

InvalidInclude – if nesting is more than MAX_INCLUDE_DEPTH.

managed_keys = ('filter', 'page', 'fields', 'sort', 'include', 'q')
property pagination: PaginationQueryStringManager

Return all page parameters as a dict.

Returns:

a dict of pagination information.

To allow multiples strategies, all parameters starting with page will be included. e.g:

{
    "number": '25',
    "size": '150',
}

Example with number strategy:

query_string = {‘page[number]’: ‘25’, ‘page[size]’: ‘10’} parsed_query.pagination {‘number’: ‘25’, ‘size’: ‘10’}

Raises:

BadRequest – if the client is not allowed to disable pagination.

property querystring: Dict[str, str]

Return original querystring but containing only managed keys.

Returns:

dict of managed querystring parameter

fastapi_jsonapi.schema module

Base JSON:API schemas.

Pydantic (for FastAPI).

class fastapi_jsonapi.schema.BaseJSONAPIDataInSchema(*, data: BaseJSONAPIItemInSchema)

Bases: BaseModel

data: BaseJSONAPIItemInSchema
class fastapi_jsonapi.schema.BaseJSONAPIItemInSchema(*, type: str, attributes: TypeSchema, relationships: TypeSchema | None = None, id: str | None = None)

Bases: BaseJSONAPIItemSchema

Schema for post/patch method

TODO POST: optionally accept custom id for object https://jsonapi.org/format/#crud-creating-client-ids TODO PATCH: accept object id (maybe create a new separate schema)

attributes: TypeSchema
id: str | None
relationships: TypeSchema | None
class fastapi_jsonapi.schema.BaseJSONAPIItemSchema(*, type: str, attributes: dict)

Bases: BaseModel

Base JSON:API item schema.

attributes: dict
type: str
class fastapi_jsonapi.schema.BaseJSONAPIObjectSchema(*, type: str, attributes: dict, id: str)

Bases: BaseJSONAPIItemSchema

Base JSON:API object schema.

id: str
class fastapi_jsonapi.schema.BaseJSONAPIRelationshipDataToManySchema(*, data: List[BaseJSONAPIRelationshipSchema])

Bases: BaseModel

data: List[BaseJSONAPIRelationshipSchema]
class fastapi_jsonapi.schema.BaseJSONAPIRelationshipDataToOneSchema(*, data: BaseJSONAPIRelationshipSchema)

Bases: BaseModel

data: BaseJSONAPIRelationshipSchema
class fastapi_jsonapi.schema.BaseJSONAPIRelationshipSchema(*, id: str, type: str)

Bases: BaseModel

class Config

Bases: BaseConfig

extra = 'forbid'
id: str
type: str
class fastapi_jsonapi.schema.BaseJSONAPIResultSchema(*, meta: JSONAPIResultListMetaSchema | None = None, jsonapi: JSONAPIDocumentObjectSchema = JSONAPIDocumentObjectSchema(version='1.0'))

Bases: BaseModel

JSON:API Required fields schema

jsonapi: JSONAPIDocumentObjectSchema
meta: JSONAPIResultListMetaSchema | None
class fastapi_jsonapi.schema.JSONAPIDocumentObjectSchema(*, version: str = '1.0')

Bases: BaseModel

JSON:API Document Object Schema.

https://jsonapi.org/format/#document-jsonapi-object

version: str
class fastapi_jsonapi.schema.JSONAPIObjectSchema(*, type: str, attributes: dict, id: str)

Bases: BaseJSONAPIObjectSchema

JSON:API base object schema.

class fastapi_jsonapi.schema.JSONAPIResultDetailSchema(*, meta: JSONAPIResultListMetaSchema | None = None, jsonapi: JSONAPIDocumentObjectSchema = JSONAPIDocumentObjectSchema(version='1.0'), data: JSONAPIObjectSchema)

Bases: BaseJSONAPIResultSchema

JSON:API base detail schema.

data: JSONAPIObjectSchema
class fastapi_jsonapi.schema.JSONAPIResultListMetaSchema(*, count: int | None = None, totalPages: int | None = None)

Bases: BaseModel

JSON:API list meta schema.

class Config

Bases: object

allow_population_by_field_name = True
count: int | None
total_pages: int | None
class fastapi_jsonapi.schema.JSONAPIResultListSchema(*, meta: JSONAPIResultListMetaSchema | None = None, jsonapi: JSONAPIDocumentObjectSchema = JSONAPIDocumentObjectSchema(version='1.0'), data: Sequence[JSONAPIObjectSchema])

Bases: BaseJSONAPIResultSchema

JSON:API list base result schema.

data: Sequence[JSONAPIObjectSchema]
exception fastapi_jsonapi.schema.JSONAPISchemaIntrospectionError

Bases: Exception

fastapi_jsonapi.schema.get_model_field(schema: Type[TypeSchema], field: str) → str

Get the model field of a schema field.

# todo: use alias (custom names)?

For example:

class Computer(sqla_base):

user = relationship(User)

class ComputerSchema(pydantic_base):

owner = Field(alias=”user”, relationship=…)

Parameters:

  • schema – a pydantic schema

  • field – the name of the schema field

Returns:

the name of the field in the model

Raises:

Exception – if the schema from parameter has no attribute for parameter.

Retrieve the related schema of a relationship field.

Params schema:

the schema to retrieve le relationship field from

Params field:

the relationship field

Returns:

the related schema

fastapi_jsonapi.schema.get_relationships(schema: Type[TypeSchema], model_field: bool = False) → List[str]

Return relationship fields of a schema.

Parameters:

  • schema – a schemas schema

  • model_field – list of relationship fields of a schema

fastapi_jsonapi.schema.get_schema_from_type(resource_type: str, app: FastAPI) → Type[BaseModel]

Retrieve a schema from the registry by his type.

Parameters:

  • resource_type – the type of the resource.

  • app – FastAPI app instance.

Return Schema:

the schema class.

Raises:

Exception – if the schema not found for this resource type.

fastapi_jsonapi.signature module

Functions for extracting and updating signatures.

fastapi_jsonapi.signature.create_additional_query_params(schema: Type[BaseModel] | None) → tuple[list[Parameter], list[Parameter]]
fastapi_jsonapi.signature.create_filter_parameter(name: str, field: ModelField) → Parameter

fastapi_jsonapi.splitter module

Splitter for filters, sorts and includes.