"""JSON API schemas builder class."""
import logging
from typing import Annotated, Any, Callable, Literal, Optional, Type, TypeVar, Union
from pydantic import AfterValidator, BeforeValidator, ConfigDict, PlainValidator, WrapValidator, create_model
from pydantic import BaseModel as PydanticBaseModel
# noinspection PyProtectedMember
from pydantic.fields import FieldInfo
from fastapi_jsonapi.common import get_relationship_info_from_field_metadata, search_client_can_set_id
from fastapi_jsonapi.schema import (
BaseJSONAPIDataInSchema,
BaseJSONAPIItemInSchema,
BaseJSONAPIRelationshipDataToManySchema,
BaseJSONAPIRelationshipDataToOneSchema,
BaseJSONAPIRelationshipSchema,
BaseJSONAPIResultSchema,
BuiltSchemasDTO,
JSONAPIObjectSchema,
JSONAPIObjectSchemas,
JSONAPIResultDetailSchema,
JSONAPIResultListSchema,
RelationshipInfoSchema,
SchemasInfoDTO,
get_schema_from_field_annotation,
)
from fastapi_jsonapi.schema_base import BaseModel, Field, registry
from fastapi_jsonapi.storages.schemas_storage import schemas_storage
from fastapi_jsonapi.types_metadata import RelationshipInfo
from fastapi_jsonapi.validation_utils import extract_validators
log = logging.getLogger(__name__)
JSONAPIObjectSchemaType = TypeVar("JSONAPIObjectSchemaType", bound=PydanticBaseModel)
class SchemaBuilder:
def __init__(
self,
resource_type: str,
):
self._resource_type = resource_type
def _create_schemas_objects_list(self, schema: Type[BaseModel]) -> Type[JSONAPIResultListSchema]:
object_jsonapi_list_schema, list_jsonapi_schema = self.build_list_schemas(schema)
# TODO: do we need this `object_jsonapi_list_schema` field? it's not used anywhere
# self.object_jsonapi_list_schema: Type[JSONAPIObjectSchema] = object_jsonapi_list_schema
return list_jsonapi_schema
def _create_schemas_object_detail(self, schema: Type[BaseModel]) -> Type[JSONAPIResultDetailSchema]:
object_jsonapi_detail_schema, detail_jsonapi_schema = self.build_detail_schemas(schema)
# TODO: do we need this `object_jsonapi_detail_schema` field? it's not used anywhere
# self.object_jsonapi_detail_schema: Type[JSONAPIObjectSchema] = object_jsonapi_detail_schema
return detail_jsonapi_schema
def create_schemas(
self,
schema: Type[BaseModel],
schema_in_post: Optional[Type[BaseModel]] = None,
schema_in_patch: Optional[Type[BaseModel]] = None,
) -> BuiltSchemasDTO:
schema_in_post = schema_in_post or schema
schema_name_in_post_suffix = ""
if any(schema_in_post is cmp_schema for cmp_schema in [schema, schema_in_patch]):
schema_name_in_post_suffix = "InPost"
schema_in_patch = schema_in_patch or schema
schema_name_in_patch_suffix = ""
if any(schema_in_patch is cmp_schema for cmp_schema in [schema, schema_in_post]):
schema_name_in_patch_suffix = "InPatch"
schema_in_post, schema_in_post_data = self.build_schema_in(
schema_in=schema_in_post,
schema=schema,
operation_type="create",
schema_name_suffix=schema_name_in_post_suffix,
non_optional_relationships=True,
)
schema_in_patch, schema_in_patch_data = self.build_schema_in(
schema_in=schema_in_patch,
schema=schema,
operation_type="update",
schema_name_suffix=schema_name_in_patch_suffix,
id_field_required=True,
)
return BuiltSchemasDTO(
schema_in_post=schema_in_post,
schema_in_post_data=schema_in_post_data,
schema_in_patch=schema_in_patch,
schema_in_patch_data=schema_in_patch_data,
list_response_schema=self._create_schemas_objects_list(schema),
detail_response_schema=self._create_schemas_object_detail(schema),
)
def build_schema_in(
self,
schema_in: Type[BaseModel],
schema,
operation_type: Literal["create", "update", "get"],
schema_name_suffix: str = "",
non_optional_relationships: bool = False,
id_field_required: bool = False,
) -> tuple[Type[BaseJSONAPIDataInSchema], Type[BaseJSONAPIItemInSchema]]:
base_schema_name = schema_in.__name__.removesuffix("Schema") + schema_name_suffix
dto = self.get_info_from_schema_for_building(
base_name=base_schema_name,
schema=schema_in,
operation_type=operation_type,
non_optional_relationships=non_optional_relationships,
)
object_jsonapi_schema = self.build_jsonapi_object(
base_name=base_schema_name,
resource_type=self._resource_type,
dto=dto,
model_base=BaseJSONAPIItemInSchema,
id_field_required=id_field_required,
)
wrapped_object_jsonapi_schema = create_model(
f"{base_schema_name}ObjectDataJSONAPI",
data=(object_jsonapi_schema, ...),
__base__=BaseJSONAPIDataInSchema,
)
schemas_storage.add_resource(
builder=self,
resource_type=self._resource_type,
operation_type=operation_type,
source_schema=schema,
data_schema=object_jsonapi_schema,
attributes_schema=dto.attributes_schema,
field_schemas=dto.field_schemas,
relationships_info=dto.relationships_info,
model_validators=dto.model_validators,
schema_in=wrapped_object_jsonapi_schema,
)
return wrapped_object_jsonapi_schema, object_jsonapi_schema
def _build_schema(
self,
base_name: str,
schema: Type[BaseModel],
builder: Callable,
):
object_schemas = self.create_jsonapi_object_schemas(
schema=schema,
base_name=base_name,
compute_included_schemas=True,
)
object_jsonapi_schema = object_schemas.object_jsonapi_schema
response_jsonapi_schema = builder(
name=base_name,
object_jsonapi_schema=object_jsonapi_schema,
includes_schemas=object_schemas.included_schemas_list,
)
return object_jsonapi_schema, response_jsonapi_schema
def build_detail_schemas(
self,
schema: Type[BaseModel],
) -> tuple[Type[JSONAPIObjectSchema], Type[JSONAPIResultDetailSchema]]:
return self._build_schema(
base_name=f"{schema.__name__}Detail",
schema=schema,
builder=self.build_schema_for_detail_result,
)
def build_list_schemas(
self,
schema: Type[BaseModel],
) -> tuple[Type[JSONAPIObjectSchema], Type[JSONAPIResultListSchema]]:
return self._build_schema(
base_name=f"{schema.__name__}List",
schema=schema,
builder=self.build_schema_for_list_result,
)
@classmethod
def _annotation_with_validators(cls, field: FieldInfo) -> type:
annotation = field.annotation
validators = []
for val in field.metadata:
if isinstance(val, (AfterValidator, BeforeValidator, WrapValidator, PlainValidator)):
validators.append(val)
if validators:
annotation = Annotated.__class_getitem__((annotation, *validators))
return annotation
def get_info_from_schema_for_building(
self,
base_name: str,
schema: Type[BaseModel],
operation_type: Literal["create", "update", "get"],
non_optional_relationships: bool = False,
) -> SchemasInfoDTO:
attributes_schema_fields = {}
relationships_schema_fields = {}
relationships_info: dict[str, tuple[RelationshipInfo, Any]] = {}
included_schemas: list[tuple[str, BaseModel, str]] = []
has_required_relationship = False
resource_id_field = (str, Field(default=None), None, {})
# required! otherwise we get ForwardRef
schema.model_rebuild(_types_namespace=registry.schemas)
for name, field in (schema.model_fields or {}).items():
if relationship_info := get_relationship_info_from_field_metadata(field):
relationships_info[name] = (relationship_info, field)
relationship_schema = self.create_relationship_data_schema(
field_name=name,
base_name=base_name,
field=field,
operation_type=operation_type,
relationship_info=relationship_info,
)
field_marked_required = field.is_required()
relationship_field = ... if (non_optional_relationships and field_marked_required) else None
if relationship_field is not None:
has_required_relationship = True
relationships_schema_fields[name] = (relationship_schema, relationship_field)
# works both for to-one and to-many
if related_schema := get_schema_from_field_annotation(field):
included_schemas.append((name, related_schema, relationship_info.resource_type))
elif name == "id":
id_validators, _ = extract_validators(
model=schema,
include_for_field_names={"id"},
)
if not (can_set_id := search_client_can_set_id.first(field)):
continue
resource_id_field = (str, can_set_id, self._annotation_with_validators(field=field), id_validators)
else:
attributes_schema_fields[name] = (self._annotation_with_validators(field=field), field.default)
model_config = ConfigDict(
from_attributes=True,
)
field_validators, model_validators = extract_validators(schema, exclude_for_field_names={"id"})
attributes_schema = create_model(
f"{base_name}AttributesJSONAPI",
**attributes_schema_fields,
__config__=model_config,
__validators__={**field_validators, **model_validators},
)
field_schemas = {}
for field_name, field in attributes_schema_fields.items():
field_validators, _ = extract_validators(
schema,
include_for_field_names={field_name},
)
field_schemas[field_name] = create_model(
f"{base_name}{field_name.title()}AttributeJSONAPI",
**{field_name: field},
__config__=model_config,
__validators__=field_validators,
)
relationships_schema = create_model(
f"{base_name}RelationshipsJSONAPI",
**relationships_schema_fields,
__config__=model_config,
)
return SchemasInfoDTO(
resource_id_field=resource_id_field,
attributes_schema=attributes_schema,
relationships_schema=relationships_schema,
relationships_info=relationships_info,
has_required_relationship=has_required_relationship,
included_schemas=included_schemas,
field_schemas=field_schemas,
model_validators=model_validators,
)
@classmethod
def create_relationship_schema(
cls,
name: str,
relationship_info: RelationshipInfo,
) -> Type[BaseJSONAPIRelationshipSchema]:
# TODO: cache?
if name.endswith("s"):
# plural to single
name = name[:-1]
return create_model(
f"{name}RelationshipJSONAPI",
id=(
str,
Field(
...,
description="Resource object id",
json_schema_extra={"example": relationship_info.resource_id_example},
),
),
type=(
str,
Field(
default=relationship_info.resource_type,
description="Resource type",
),
),
__base__=BaseJSONAPIRelationshipSchema,
)
def create_relationship_data_schema(
self,
field_name: str,
base_name: str,
operation_type: Literal["create", "update", "get"],
field: FieldInfo,
relationship_info: RelationshipInfo,
) -> RelationshipInfoSchema:
if relationship_schema := schemas_storage.get_relationship_schema(
from_resource_type=self._resource_type,
to_resource_type=relationship_info.resource_type,
operation_type=operation_type,
field_name=field_name,
):
return relationship_schema
base_name = base_name.removesuffix("Schema")
schema_name = f"{base_name}{field_name.title()}"
relationship_schema = self.create_relationship_schema(
name=schema_name,
relationship_info=relationship_info,
)
base = BaseJSONAPIRelationshipDataToOneSchema
if relationship_info.many:
relationship_schema = list[relationship_schema]
base = BaseJSONAPIRelationshipDataToManySchema
elif not field.is_required():
relationship_schema = Optional[relationship_schema]
relationship_data_schema = create_model(
f"{schema_name}RelationshipDataJSONAPI",
# TODO: on create (post request) sometimes it's required and at the same time on fetch it's not required
data=(relationship_schema, Field(... if field.is_required() else None)),
__base__=base,
)
schemas_storage.add_relationship(
from_resource_type=self._resource_type,
to_resource_type=relationship_info.resource_type,
operation_type=operation_type,
field_name=field_name,
relationship_schema=relationship_data_schema,
relationship_info=relationship_info,
)
return relationship_data_schema
def build_jsonapi_object(
self,
base_name: str,
resource_type: str,
dto: SchemasInfoDTO,
model_base: Type[JSONAPIObjectSchemaType] = JSONAPIObjectSchema,
with_relationships: bool = True,
id_field_required: bool = False,
) -> Type[JSONAPIObjectSchemaType]:
field_type, can_set_id, id_cast_func, id_validators = dto.resource_id_field
if can_set_id:
field_type = Annotated[field_type, can_set_id]
relationship_less_fields = {}
relationship_less_fields.update(
id=(field_type, Field(... if id_field_required else None)),
attributes=(dto.attributes_schema, ...),
type=(str, Field(default=resource_type or self._resource_type, description="Resource type")),
)
object_jsonapi_schema_fields = {}
object_jsonapi_schema_fields.update(
id=(field_type, Field(... if id_field_required else None)),
attributes=(dto.attributes_schema, ...),
type=(str, Field(default=resource_type or self._resource_type, description="Resource type")),
)
if with_relationships:
object_jsonapi_schema_fields.update(
relationships=(Optional[dto.relationships_schema], ... if dto.has_required_relationship else None),
)
object_jsonapi_schema = create_model(
f"{base_name}ObjectJSONAPI",
**object_jsonapi_schema_fields,
__validators__=id_validators,
__base__=model_base,
)
return object_jsonapi_schema
def find_all_included_schemas(
self,
included_schemas: list[tuple[str, BaseModel, str]],
) -> dict[str, Type[JSONAPIObjectSchema]]:
return {
name: self.create_jsonapi_object_schemas(
included_schema,
resource_type=resource_type,
).object_jsonapi_schema
for (name, included_schema, resource_type) in included_schemas
}
def create_jsonapi_object_schemas(
self,
schema: Type[BaseModel],
resource_type: Optional[str] = None,
base_name: str = "",
compute_included_schemas: bool = False,
) -> JSONAPIObjectSchemas:
resource_type = resource_type or self._resource_type
if object_schema := schemas_storage.get_jsonapi_object_schema(
source_schema=schema,
resource_type=resource_type,
):
return object_schema
base_name = base_name or schema.__name__
dto = self.get_info_from_schema_for_building(
base_name=base_name,
operation_type="get",
schema=schema,
)
object_jsonapi_schema = self.build_jsonapi_object(
base_name=base_name,
resource_type=resource_type or self._resource_type,
dto=dto,
)
relationship_less_object_jsonapi_schema = self.build_jsonapi_object(
base_name=base_name,
resource_type=resource_type or self._resource_type,
dto=dto,
with_relationships=False,
)
schemas_storage.add_resource(
builder=self,
resource_type=self._resource_type,
operation_type="get",
source_schema=schema,
data_schema=relationship_less_object_jsonapi_schema,
attributes_schema=dto.attributes_schema,
field_schemas=dto.field_schemas,
relationships_info=dto.relationships_info,
model_validators=dto.model_validators,
)
can_be_included_schemas = {}
if compute_included_schemas:
can_be_included_schemas = self.find_all_included_schemas(included_schemas=dto.included_schemas)
result = JSONAPIObjectSchemas(
attributes_schema=dto.attributes_schema,
relationships_schema=dto.relationships_schema,
object_jsonapi_schema=object_jsonapi_schema,
can_be_included_schemas=can_be_included_schemas,
)
schemas_storage.add_jsonapi_object_schema(
source_schema=schema,
resource_type=resource_type,
jsonapi_object_schema=result,
)
return result
def build_schema_for_list_result(
self,
name: str,
object_jsonapi_schema: Type[JSONAPIObjectSchema],
includes_schemas: list[Type[JSONAPIObjectSchema]],
) -> Type[JSONAPIResultListSchema]:
return self.build_schema_for_result(
name=f"{name}JSONAPI",
base=JSONAPIResultListSchema,
data_type=list[object_jsonapi_schema],
includes_schemas=includes_schemas,
)
def build_schema_for_detail_result(
self,
name: str,
object_jsonapi_schema: Type[JSONAPIObjectSchema],
includes_schemas: list[Type[JSONAPIObjectSchema]],
) -> Type[JSONAPIResultDetailSchema]:
return self.build_schema_for_result(
name=f"{name}JSONAPI",
base=JSONAPIResultDetailSchema,
data_type=object_jsonapi_schema,
includes_schemas=includes_schemas,
)
@classmethod
def build_schema_for_result(
cls,
name: str,
base: Type[BaseJSONAPIResultSchema],
data_type: Union[Type[JSONAPIObjectSchema], Type[list[JSONAPIObjectSchema]]],
includes_schemas: list[Type[JSONAPIObjectSchema]],
) -> Union[Type[JSONAPIResultListSchema], Type[JSONAPIResultDetailSchema]]:
included_schema_annotation = Union[JSONAPIObjectSchema]
for includes_schema in includes_schemas:
included_schema_annotation = Union[included_schema_annotation, includes_schema]
schema_fields = {
"data": (data_type, ...),
}
if includes_schemas:
schema_fields.update(
included=(
list[included_schema_annotation],
Field(default=None),
),
)
return create_model(
name,
**schema_fields,
__base__=base,
)