### Install apischema with examples extra Source: https://wyfo.github.io/apischema Install apischema along with its example dependencies, including libraries like SQLAlchemy or attrs, using the `examples` extra. This is necessary to run the documentation examples. ```bash pip install apischema[examples] ``` -------------------------------- ### Install apischema Source: https://wyfo.github.io/apischema Install the apischema library using pip. It requires Python 3.8+. ```bash pip install apischema ``` -------------------------------- ### Using Relay Connections Source: https://wyfo.github.io/apischema/0.19/graphql/relay Example demonstrating the implementation of a standard Relay connection with a resolver. ```python from dataclasses import dataclass from typing import Optional, TypeVar import graphql from graphql.utilities import print_schema from apischema.graphql import graphql_schema, relay, resolver Cursor = int # let's use an integer cursor in all our connection Node = TypeVar("Node") Connection = relay.Connection[Node, Cursor, relay.Edge[Node, Cursor]] # Connection can now be used just like Connection[Ship] or Connection[Faction | None] @dataclass class Ship: name: str @dataclass class Faction: @resolver def ships( self, first: int | None, after: Cursor | None ) -> Connection[Optional[Ship]] | None: edges = [relay.Edge(Ship("X-Wing"), 0), relay.Edge(Ship("B-Wing"), 1)] return Connection(edges, relay.PageInfo.from_edges(edges)) def faction() -> Faction | None: return Faction() schema = graphql_schema(query=[faction]) schema_str = """\ type Query { faction: Faction } type Faction { ships(first: Int, after: Int): ShipConnection } type ShipConnection { edges: [ShipEdge] pageInfo: PageInfo! } type ShipEdge { node: Ship cursor: Int! } type Ship { name: String! } type PageInfo { hasPreviousPage: Boolean! hasNextPage: Boolean! startCursor: Int endCursor: Int }""" assert print_schema(schema) == schema_str query = """ { faction { ships { pageInfo { endCursor hasNextPage } edges { cursor node { name } } } } }""" assert graphql.graphql_sync(schema, query).data == { "faction": { "ships": { "pageInfo": {"endCursor": 1, "hasNextPage": False}, "edges": [ {"cursor": 0, "node": {"name": "X-Wing"}}, {"cursor": 1, "node": {"name": "B-Wing"}}, ], } } } ``` -------------------------------- ### Install apischema with GraphQL support Source: https://wyfo.github.io/apischema/0.19/graphql/overview Install apischema with the graphql extra requirement to enable GraphQL support. ```bash pip install apischema[graphql] ``` -------------------------------- ### Customizing Connections and Edges Source: https://wyfo.github.io/apischema/0.19/graphql/relay Example showing how to subclass relay.Connection and relay.Edge to add custom fields. ```python from dataclasses import dataclass from typing import Optional, TypeVar from graphql import print_schema from apischema.graphql import graphql_schema, relay, resolver Cursor = int Node = TypeVar("Node") Edge = TypeVar("Edge", bound=relay.Edge) @dataclass class MyConnection(relay.Connection[Node, Cursor, Edge]): connection_field: bool @dataclass class MyEdge(relay.Edge[Node, Cursor]): edge_field: int | None Connection = MyConnection[Node, MyEdge[Node]] @dataclass class Ship: name: str @dataclass class Faction: @resolver def ships( self, first: int | None, after: Cursor | None ) -> Connection[Optional[Ship]] | None: ... def faction() -> Faction | None: return Faction() schema = graphql_schema(query=[faction]) schema_str = """\ type Query { faction: Faction } type Faction { ships(first: Int, after: Int): ShipConnection } type ShipConnection { edges: [ShipEdge] pageInfo: PageInfo! connectionField: Boolean! } type ShipEdge { node: Ship cursor: Int! edgeField: Int } type Ship { name: String! } type PageInfo { hasPreviousPage: Boolean! hasNextPage: Boolean! startCursor: Int endCursor: Int }""" assert print_schema(schema) == schema_str ``` -------------------------------- ### GraphQL Query Example Source: https://wyfo.github.io/apischema/0.19/examples/subclass_tagged_union A GraphQL query to echo drawing data, demonstrating the use of nested input types like ConcatInput. ```graphql query { echo(drawing: { Concat: { left: { SizedLine: { start: 0, stop: 12, size: 3 }, }, right: { Line: { start: 12, stop: 13 }, } } }) { Concat { left { Line { start stop step } } right { Line { start stop step } } } } } ``` -------------------------------- ### Serializer Inheritance Example Source: https://wyfo.github.io/apischema/0.19/conversions Demonstrates how serializers are inherited by default and can be overridden in subclasses. This applies to both standalone serializer functions and methods decorated with @serializer. ```python from apischema import serialize, serializer class Foo: pass @serializer def serialize_foo(foo: Foo) -> int: return 0 class Foo2(Foo): pass # Deserializer is inherited assert serialize(Foo, Foo()) == serialize(Foo2, Foo2()) == 0 class Bar: @serializer def serialize(self) -> int: return 0 class Bar2(Bar): def serialize(self) -> int: return 1 # Deserializer is inherited and overridden assert serialize(Bar, Bar()) == 0 != serialize(Bar2, Bar2()) == 1 ``` -------------------------------- ### Define GraphQL Schema with Subscriptions Source: https://wyfo.github.io/apischema/0.19/graphql/schema Defines a GraphQL schema including a subscription field that yields events. Ensure `graphql` and `apischema` are installed. ```python import asyncio from typing import AsyncIterable import graphql from graphql import print_schema from apischema.graphql import graphql_schema def hello() -> str: return "world" async def events() -> AsyncIterable[str]: yield "bonjour" yield "au revoir" schema = graphql_schema(query=[hello], subscription=[events]) schema_str = """ type Query { hello: String! } type Subscription { events: String! } """ assert print_schema(schema) == schema_str async def test(): subscription = await graphql.subscribe( schema, graphql.parse("subscription {events}") ) assert [event.data async for event in subscription] == [ {"events": "bonjour"}, {"events": "au revoir"}, ] asyncio.run(test()) ``` -------------------------------- ### Precompute (De)serialization Methods Source: https://wyfo.github.io/apischema/0.19/optimizations_and_benchmark Use `deserialization_method` and `serialization_method` to get precomputed methods for faster (de)serialization. These methods are cached using `functools.lru_cache` and can increase performance by 10% when used directly. ```python from dataclasses import dataclass from apischema import deserialization_method, serialization_method @dataclass class Foo: bar: int deserialize_foo = deserialization_method(Foo) serialize_foo = serialization_method(Foo) assert deserialize_foo({"bar": 0}) == Foo(0) assert serialize_foo(Foo(0)) == {"bar": 0} ``` -------------------------------- ### Implement Sub-conversions for Dynamic Serialization Source: https://wyfo.github.io/apischema/0.19/conversions Utilize `sub_conversion` within a `Conversion` object to apply a secondary conversion to the result of a primary conversion. This example demonstrates serializing a `Query` type. ```python from dataclasses import dataclass from typing import Generic, TypeVar from apischema.conversions import Conversion from apischema.json_schema import serialization_schema T = TypeVar("T") class Query(Generic[T]): ... def query_to_list(q: Query[T]) -> list[T]: ... def query_to_scalar(q: Query[T]) -> T | None: ... @dataclass class FooModel: bar: int class Foo: def serialize(self) -> FooModel: ... assert serialization_schema( Query[Foo], conversion=Conversion(query_to_list, sub_conversion=Foo.serialize) ) == { # We get an array of Foo "type": "array", "items": { "type": "object", "properties": {"bar": {"type": "integer"}}, "required": ["bar"], "additionalProperties": False, }, "$schema": "http://json-schema.org/draft/2020-12/schema#", } ``` -------------------------------- ### Generic Conversions with Wrapper Source: https://wyfo.github.io/apischema/0.19/conversions Illustrates generic conversions using TypeVars and a Wrapper class. The @serializer decorator and deserializer function are used to handle generic types, and examples show deserialization, validation, and schema generation. ```python from typing import Generic, TypeVar import pytest from apischema import ValidationError, deserialize, serialize from apischema.conversions import deserializer, serializer from apischema.json_schema import deserialization_schema, serialization_schema T = TypeVar("T") class Wrapper(Generic[T]): def __init__(self, wrapped: T): self.wrapped = wrapped @serializer def unwrap(self) -> T: return self.wrapped # Wrapper constructor can be used as a function too (so deserializer could work as decorator) deserializer(Wrapper) assert deserialize(Wrapper[list[int]], [0, 1]).wrapped == [0, 1] with pytest.raises(ValidationError): deserialize(Wrapper[int], "wrapped") assert serialize(Wrapper[str], Wrapper("wrapped")) == "wrapped" assert ( deserialization_schema(Wrapper[int]) == {"$schema": "http://json-schema.org/draft/2020-12/schema#", "type": "integer"} == serialization_schema(Wrapper[int]) ) ``` -------------------------------- ### Implement a Line Drawing Source: https://wyfo.github.io/apischema/0.19/examples/subclass_tagged_union The `Line` class inherits from `Drawing` and implements the `points` method to yield points based on start, stop, and step values. It includes metadata for schema generation, such as `exc_min=0` for the step. ```python @dataclass class Line(Drawing): start: float stop: float step: float = field(default=1, metadata=schema(exc_min=0)) async def points(self) -> AsyncIterable[float]: point = self.start while point <= self.stop: yield point point += self.step ``` -------------------------------- ### Register Deserializers for External Types and Subclasses Source: https://wyfo.github.io/apischema/0.19/examples/inherited_deserializer Register deserializers for external types and their subclasses. This example defines `ForeignType` and `ForeignSubtype`, then uses a recursive function `rec_subclasses` to find all subclasses. Deserializers are registered for each subclass, allowing `apischema` to deserialize integers into instances of these types. ```python # For external types (defines in imported library) @dataclass class ForeignType: value: int class ForeignSubtype(ForeignType): pass T = TypeVar("T") # Recursive implementation of type.__subclasses__ def rec_subclasses(cls: type[T]) -> Iterator[type[T]]: for sub_cls in cls.__subclasses__(): yield sub_cls yield from rec_subclasses(sub_cls) # Register deserializers for all subclasses for cls in (ForeignType, *rec_subclasses(ForeignType)): # cls=cls is an lambda idiom to capture variable by value inside loop deserializer(Conversion(lambda value, cls=cls: cls(value), source=int, target=cls)) assert deserialize(ForeignType, 0) == ForeignType(0) assert deserialize(ForeignSubtype, 0) == ForeignSubtype(0) ``` -------------------------------- ### Overview of apischema functionality Source: https://wyfo.github.io/apischema/0.19/de_serialization Demonstrates basic deserialization, serialization, validation, JSON Schema generation, and GraphQL schema creation using dataclasses. ```python from collections.abc import Collection from dataclasses import dataclass, field from uuid import UUID, uuid4 import pytest from graphql import print_schema from apischema import ValidationError, deserialize, serialize from apischema.graphql import graphql_schema from apischema.json_schema import deserialization_schema # Define a schema with standard dataclasses @dataclass class Resource: id: UUID name: str tags: set[str] = field(default_factory=set) # Get some data uuid = uuid4() data = {"id": str(uuid), "name": "wyfo", "tags": ["some_tag"]} # Deserialize data resource = deserialize(Resource, data) assert resource == Resource(uuid, "wyfo", {"some_tag"}) # Serialize objects assert serialize(Resource, resource) == data # Validate during deserialization with pytest.raises(ValidationError) as err: # pytest checks exception is raised deserialize(Resource, {"id": "42", "name": "wyfo"}) assert err.value.errors == [ {"loc": ["id"], "err": "badly formed hexadecimal UUID string"} ] # Generate JSON Schema assert deserialization_schema(Resource) == { "$schema": "http://json-schema.org/draft/2020-12/schema#", "type": "object", "properties": { "id": {"type": "string", "format": "uuid"}, "name": {"type": "string"}, "tags": { "type": "array", "items": {"type": "string"}, "uniqueItems": True, "default": [], }, }, "required": ["id", "name"], "additionalProperties": False, } # Define GraphQL operations def resources(tags: Collection[str] | None = None) -> Collection[Resource] | None: ... # Generate GraphQL schema schema = graphql_schema(query=[resources], id_types={UUID}) schema_str = """\ type Query { resources(tags: [String!]): [Resource!] } type Resource { id: ID! name: String! tags: [String!]! }""" assert print_schema(schema) == schema_str ``` -------------------------------- ### Python GraphQL Execution Source: https://wyfo.github.io/apischema/0.19/examples/subclass_tagged_union Python code demonstrating how to execute a GraphQL query against a schema using graphql-core. ```python assert graphql.graphql_sync(drawing_schema, query).data == { "echo": { "Concat": { "left": {"Line": {"start": 0.0, "stop": 12.0, "step": 6.0}}, "right": {"Line": {"start": 12.0, "stop": 13.0, "step": 1.0}}, } } } ``` -------------------------------- ### Dynamic Conversions at Runtime Source: https://wyfo.github.io/apischema/0.19/conversions Demonstrates providing conversions dynamically at runtime using the 'conversion' parameter in deserialize/serialize functions. This allows specifying custom conversion logic for specific calls, including using Annotated metadata. ```python import os import time from dataclasses import dataclass from datetime import datetime from typing import Annotated from apischema import deserialize, serialize from apischema.metadata import conversion # Set UTC timezone for example os.environ["TZ"] = "UTC" time.tzset() def datetime_from_timestamp(timestamp: int) -> datetime: return datetime.fromtimestamp(timestamp) date = datetime(2017, 9, 2) assert deserialize(datetime, 1504310400, conversion=datetime_from_timestamp) == date @dataclass class Foo: bar: int baz: int def sum(self) -> int: return self.bar + self.baz @property def diff(self) -> int: return int(self.bar - self.baz) assert serialize(Foo, Foo(0, 1)) == {"bar": 0, "baz": 1} assert serialize(Foo, Foo(0, 1), conversion=Foo.sum) == 1 assert serialize(Foo, Foo(0, 1), conversion=Foo.diff) == -1 # conversions can be specified using Annotated assert serialize(Annotated[Foo, conversion(serialization=Foo.sum)], Foo(0, 1)) == 1 ``` -------------------------------- ### Custom ID Encoding with Base64 Source: https://wyfo.github.io/apischema/0.19/graphql/data_model_and_resolvers Illustrates how to implement custom serialization and deserialization for the ID type using base64 encoding. This example maps a UUID to a base64 encoded string in the GraphQL schema. ```python from base64 import b64decode, b64encode from dataclasses import dataclass from uuid import UUID from graphql import graphql_sync from apischema.graphql import graphql_schema @dataclass class Foo: id: UUID def foo() -> Foo | None: return Foo(UUID("58c88e87-5769-4723-8974-f9ec5007a38b")) schema = graphql_schema( query=[foo], id_types={UUID}, id_encoding=( lambda s: b64decode(s).decode(), lambda s: b64encode(s.encode()).decode(), ), ) assert graphql_sync(schema, "{foo{id}}").data == { "foo": {"id": "NThjODhlODctNTc2OS00NzIzLTg5NzQtZjllYzUwMDdhMzhi"} } ``` -------------------------------- ### Define base schema settings Source: https://wyfo.github.io/apischema/0.19/json_schema Configures global schema generation for types, fields, and methods using docstring parsing. ```python from dataclasses import dataclass, field from typing import Any, Callable, get_origin import docstring_parser from apischema import schema, serialized, settings from apischema.json_schema import serialization_schema from apischema.schemas import Schema from apischema.type_names import get_type_name @dataclass class Foo: """Foo class :var bar: bar attribute""" bar: str = field(metadata=schema(max_len=10)) @serialized @property def baz(self) -> int: """baz method""" ... def type_base_schema(tp: Any) -> Schema | None: if not hasattr(tp, "__doc__"): return None return schema( title=get_type_name(tp).json_schema, description=docstring_parser.parse(tp.__doc__).short_description, ) def field_base_schema(tp: Any, name: str, alias: str) -> Schema | None: title = alias.replace("_", " ").capitalize() tp = get_origin(tp) or tp # tp can be generic for meta in docstring_parser.parse(tp.__doc__).meta: if meta.args == ["var", name]: return schema(title=title, description=meta.description) return schema(title=title) def method_base_schema(tp: Any, method: Callable, alias: str) -> Schema | None: return schema( title=alias.replace("_", " ").capitalize(), description=docstring_parser.parse(method.__doc__).short_description, ) settings.base_schema.type = type_base_schema settings.base_schema.field = field_base_schema settings.base_schema.method = method_base_schema assert serialization_schema(Foo) == { "$schema": "http://json-schema.org/draft/2020-12/schema#", "additionalProperties": False, "title": "Foo", "description": "Foo class", "properties": { "bar": { "description": "bar attribute", "title": "Bar", "type": "string", "maxLength": 10, }, "baz": {"description": "baz method", "title": "Baz", "type": "integer"}, }, "required": ["bar", "baz"], "type": "object", } ``` -------------------------------- ### Import field tracking utilities Source: https://wyfo.github.io/apischema/0.19/de_serialization Import helpers for managing field set metadata on dataclasses. ```python from dataclasses import dataclass from apischema import deserialize from apischema.fields import ( fields_set, is_set, set_fields, unset_fields, with_fields_set, ) ``` -------------------------------- ### Specify JSON Schema and OpenAPI Versions Source: https://wyfo.github.io/apischema/0.19/json_schema Demonstrates how to generate schemas for different versions like Draft 7 or OpenAPI 3.x using the version parameter. ```python from dataclasses import dataclass from typing import Literal from apischema.json_schema import ( JsonSchemaVersion, definitions_schema, deserialization_schema, ) @dataclass class Bar: baz: int | None constant: Literal[0] = 0 @dataclass class Foo: bar: Bar assert deserialization_schema(Foo, all_refs=True) == { "$schema": "http://json-schema.org/draft/2020-12/schema#", "$ref": "#/$defs/Foo", "$defs": { "Foo": { "type": "object", "properties": {"bar": {"$ref": "#/$defs/Bar"}}, "required": ["bar"], "additionalProperties": False, }, "Bar": { "type": "object", "properties": { "baz": {"type": ["integer", "null"]}, "constant": {"type": "integer", "const": 0, "default": 0}, }, "required": ["baz"], "additionalProperties": False, }, }, } assert deserialization_schema( Foo, all_refs=True, version=JsonSchemaVersion.DRAFT_7 ) == { "$schema": "http://json-schema.org/draft-07/schema#", # $ref is isolated in allOf + draft 7 prefix "allOf": [{"$ref": "#/definitions/Foo"}], "definitions": { # not "$defs" "Foo": { "type": "object", "properties": {"bar": {"$ref": "#/definitions/Bar"}}, "required": ["bar"], "additionalProperties": False, }, "Bar": { "type": "object", "properties": { "baz": {"type": ["integer", "null"]}, "constant": {"type": "integer", "const": 0, "default": 0}, }, "required": ["baz"], "additionalProperties": False, }, }, } assert deserialization_schema(Foo, version=JsonSchemaVersion.OPEN_API_3_1) == { # No definitions for OpenAPI, use definitions_schema for it "$ref": "#/components/schemas/Foo" # OpenAPI prefix } assert definitions_schema( deserialization=[Foo], version=JsonSchemaVersion.OPEN_API_3_1 ) == { "Foo": { "type": "object", "properties": {"bar": {"$ref": "#/components/schemas/Bar"}}, "required": ["bar"], "additionalProperties": False, }, "Bar": { "type": "object", "properties": { "baz": {"type": ["integer", "null"]}, "constant": {"type": "integer", "const": 0, "default": 0}, }, "required": ["baz"], "additionalProperties": False, }, } assert definitions_schema( deserialization=[Foo], version=JsonSchemaVersion.OPEN_API_3_0 ) == { "Foo": { "type": "object", "properties": {"bar": {"$ref": "#/components/schemas/Bar"}}, "required": ["bar"], "additionalProperties": False, }, "Bar": { "type": "object", # "nullable" instead of "type": "null" "properties": { "baz": {"type": "integer", "nullable": True}, "constant": {"type": "integer", "enum": [0], "default": 0}, }, "required": ["baz"], "additionalProperties": False, }, } ``` -------------------------------- ### Registering custom serializers and deserializers Source: https://wyfo.github.io/apischema/0.19/conversions Demonstrates using @serializer and @deserializer decorators to handle custom type conversions for an RGB class. ```python from dataclasses import dataclass from apischema import deserialize, schema, serialize from apischema.conversions import deserializer, serializer from apischema.json_schema import deserialization_schema, serialization_schema @schema(pattern=r"^#[0-9a-fA-F]{6}$") @dataclass class RGB: red: int green: int blue: int @serializer @property def hexa(self) -> str: return f"#{self.red:02x}{self.green:02x}{self.blue:02x}" # serializer can also be called with methods/properties outside of the class # For example, `serializer(RGB.hexa)` would have the same effect as the decorator above @deserializer def from_hexa(hexa: str) -> RGB: return RGB(int(hexa[1:3], 16), int(hexa[3:5], 16), int(hexa[5:7], 16)) assert deserialize(RGB, "#000000") == RGB(0, 0, 0) assert serialize(RGB, RGB(0, 0, 42)) == "#00002a" assert ( deserialization_schema(RGB) == serialization_schema(RGB) == { "$schema": "http://json-schema.org/draft/2020-12/schema#", "type": "string", "pattern": "^#[0-9a-fA-F]{6}$", } ) ``` -------------------------------- ### Handle generic dynamic conversions Source: https://wyfo.github.io/apischema/0.19/conversions Demonstrates support for generic types and partially specialized generics in dynamic conversions. ```python from collections.abc import Mapping, Sequence from operator import itemgetter from typing import TypeVar from apischema import serialize from apischema.json_schema import serialization_schema T = TypeVar("T") Priority = int def sort_by_priority(values_with_priority: Mapping[T, Priority]) -> Sequence[T]: return [k for k, _ in sorted(values_with_priority.items(), key=itemgetter(1))] assert serialize( dict[str, Priority], {"a": 1, "b": 0}, conversion=sort_by_priority ) == ["b", "a"] assert serialization_schema(dict[str, Priority], conversion=sort_by_priority) == { "$schema": "http://json-schema.org/draft/2020-12/schema#", "type": "array", "items": {"type": "string"}, } ``` -------------------------------- ### Apply dynamic conversions Source: https://wyfo.github.io/apischema/0.19/conversions Demonstrates that dynamic conversions are discarded for dataclasses but applied to container elements. ```python import os import time from dataclasses import dataclass from datetime import datetime from apischema import serialize # Set UTC timezone for example os.environ["TZ"] = "UTC" time.tzset() def to_timestamp(d: datetime) -> int: return int(d.timestamp()) @dataclass class Foo: bar: datetime # timestamp conversion is not applied on Foo field because it's discarded # when encountering Foo assert serialize(Foo, Foo(datetime(2019, 10, 13)), conversion=to_timestamp) == { "bar": "2019-10-13T00:00:00" } # timestamp conversion is applied on every member of list assert serialize(list[datetime], [datetime(1970, 1, 1)], conversion=to_timestamp) == [0] ``` -------------------------------- ### Basic apischema Usage: Dataclasses, Serialization, and Validation Source: https://wyfo.github.io/apischema Demonstrates defining a schema with dataclasses, serializing/deserializing data, and validating during deserialization. Includes JSON schema generation. Requires pytest for exception handling. ```python from collections.abc import Collection from dataclasses import dataclass, field from uuid import UUID, uuid4 import pytest from graphql import print_schema from apischema import ValidationError, deserialize, serialize from apischema.graphql import graphql_schema from apischema.json_schema import deserialization_schema # Define a schema with standard dataclasses @dataclass class Resource: id: UUID name: str tags: set[str] = field(default_factory=set) # Get some data uuid = uuid4() data = {"id": str(uuid), "name": "wyfo", "tags": ["some_tag"]} # Deserialize data resource = deserialize(Resource, data) assert resource == Resource(uuid, "wyfo", {"some_tag"}) # Serialize objects assert serialize(Resource, resource) == data # Validate during deserialization with pytest.raises(ValidationError) as err: # pytest checks exception is raised deserialize(Resource, {"id": "42", "name": "wyfo"}) assert err.value.errors == [ {"loc": ["id"], "err": "badly formed hexadecimal UUID string"} ] # Generate JSON Schema assert deserialization_schema(Resource) == { "$schema": "http://json-schema.org/draft/2020-12/schema#", "type": "object", "properties": { "id": {"type": "string", "format": "uuid"}, "name": {"type": "string"}, "tags": { "type": "array", "items": {"type": "string"}, "uniqueItems": True, "default": [], }, }, "required": ["id", "name"], "additionalProperties": False, } # Define GraphQL operations def resources(tags: Collection[str] | None = None) -> Collection[Resource] | None: ... # Generate GraphQL schema schema = graphql_schema(query=[resources], id_types={UUID}) schema_str = """ type Query { resources(tags: [String!]): [Resource!] } type Resource { id: ID! name: String! tags: [String!]! }""" assert print_schema(schema) == schema_str ``` -------------------------------- ### Implement String Conversions Source: https://wyfo.github.io/apischema/0.19/conversions Use as_str to register string-based serialization and deserialization for classes with string constructors and __str__ methods. ```python import bson import pytest from apischema import Unsupported, deserialize, serialize from apischema.conversions import as_str with pytest.raises(Unsupported): deserialize(bson.ObjectId, "0123456789ab0123456789ab") with pytest.raises(Unsupported): serialize(bson.ObjectId, bson.ObjectId("0123456789ab0123456789ab")) as_str(bson.ObjectId) assert deserialize(bson.ObjectId, "0123456789ab0123456789ab") == bson.ObjectId( "0123456789ab0123456789ab" ) assert ( serialize(bson.ObjectId, bson.ObjectId("0123456789ab0123456789ab")) == "0123456789ab0123456789ab" ) ``` -------------------------------- ### Generate GraphQL Schema and Execute Query Source: https://wyfo.github.io/apischema/0.19/graphql/overview Defines data models (User, Post), sample data, and resolver functions. Generates a GraphQL schema using `graphql_schema` and executes a query against it using `graphql_sync`. ```python from dataclasses import dataclass from datetime import date, datetime from typing import Collection from uuid import UUID, uuid4 from graphql import graphql_sync, print_schema from apischema.graphql import graphql_schema, resolver @dataclass class User: id: UUID username: str birthday: date | None = None @resolver def posts(self) -> Collection["Post"]: return [post for post in POSTS if post.author.id == self.id] @dataclass class Post: id: UUID author: User date: datetime content: str USERS = [User(uuid4(), "foo"), User(uuid4(), "bar")] POSTS = [Post(uuid4(), USERS[0], datetime.now(), "Hello world!")] def users() -> Collection[User]: return USERS def posts() -> Collection[Post]: return POSTS def user(username: str) -> User | None: for user in users(): if user.username == username: return user else: return None schema = graphql_schema(query=[users, user, posts], id_types={UUID}) schema_str = """ type Query { users: [User!]! user(username: String!): User posts: [Post!]! } type User { id: ID! username: String! birthday: Date posts: [Post!]! } scalar Date type Post { id: ID! author: User! date: Datetime! content: String! } scalar Datetime""" assert print_schema(schema) == schema_str query = """ { users { username posts { content } } } """ assert graphql_sync(schema, query).data == { "users": [ {"username": "foo", "posts": [{"content": "Hello world!"}]}, {"username": "bar", "posts": []}, ] } ``` -------------------------------- ### GraphQL Subscription with Event Generator and Resolver Source: https://wyfo.github.io/apischema/0.19/graphql/schema Implements a GraphQL subscription using both an event generator and a resolver for data transformation. The resolver processes events yielded by the generator. ```python import asyncio from dataclasses import dataclass from typing import AsyncIterable import graphql from graphql import print_schema from apischema.graphql import Subscription, graphql_schema def hello() -> str: return "world" async def events() -> AsyncIterable[str]: yield "bonjour" yield "au revoir" @dataclass class Message: body: str # Message can also be used directly as a function schema = graphql_schema( query=[hello], subscription=[Subscription(events, alias="messageReceived", resolver=Message)], ) schema_str = """ type Query { hello: String! } type Subscription { messageReceived: Message! } type Message { body: String! } """ assert print_schema(schema) == schema_str async def test(): subscription = await graphql.subscribe( schema, graphql.parse("subscription {messageReceived {body}}") ) assert [event.data async for event in subscription] == [ {"messageReceived": {"body": "bonjour"}}, {"messageReceived": {"body": "au revoir"}}, ] asyncio.run(test()) ``` -------------------------------- ### Implement Tagged Unions with apischema Source: https://wyfo.github.io/apischema/0.19/graphql/data_model_and_resolvers Demonstrates the implementation of tagged unions using `apischema.tagged_unions.TaggedUnion` and `apischema.tagged_unions.Tagged`. This allows for representing discriminated unions in a type-safe manner. ```python from dataclasses import dataclass import pytest from apischema import Undefined, ValidationError, alias, deserialize, schema, serialize from apischema.tagged_unions import Tagged, TaggedUnion, get_tagged @dataclass class Bar: field: str class Foo(TaggedUnion): bar: Tagged[Bar] # Tagged can have metadata like a dataclass fields i: Tagged[int] = Tagged(alias("baz") | schema(min=0)) # Instantiate using class fields tagGED_bar = Foo.bar(Bar("value")) # you can also use default constructor, but it's not typed-checked assert tagged_bar == Foo(bar=Bar("value")) # All fields that are not tagged are Undefined assert tagged_bar.bar is not Undefined and tagged_bar.i is Undefined # get_tagged allows to retrieve the tag and it's value # (but the value is not typed-checked) assert get_tagged(tagged_bar) == ("bar", Bar("value")) # (De)serialization works as expected assert deserialize(Foo, {"bar": {"field": "value"}}) == tagged_bar assert serialize(Foo, tagged_bar) == {"bar": {"field": "value"}} with pytest.raises(ValidationError) as err: deserialize(Foo, {"unknown": 42}) assert err.value.errors == [{"loc": ["unknown"], "err": "unexpected property"}] with pytest.raises(ValidationError) as err: deserialize(Foo, {"bar": {"field": "value"}, "baz": 0}) assert err.value.errors == [ {"loc": [], "err": "property count greater than 1 (maxProperties)"} ] ``` -------------------------------- ### Generate JSON Schema from Dataclass Source: https://wyfo.github.io/apischema/0.19/json_schema Demonstrates generating and comparing deserialization and serialization schemas for a simple dataclass. ```python from dataclasses import dataclass from apischema.json_schema import deserialization_schema, serialization_schema @dataclass class Foo: bar: str assert deserialization_schema(Foo) == serialization_schema(Foo) assert deserialization_schema(Foo) == { "$schema": "http://json-schema.org/draft/2020-12/schema#", "additionalProperties": False, "properties": {"bar": {"type": "string"}}, "required": ["bar"], "type": "object", } ``` -------------------------------- ### Custom Reference Factory for JSON Schema Source: https://wyfo.github.io/apischema/0.19/json_schema Illustrates how to provide a custom `ref_factory` function to `deserialization_schema`. This allows control over how schema references are formatted, such as prefixing them with a domain and path. ```python from dataclasses import dataclass from apischema.json_schema import deserialization_schema @dataclass class Foo: bar: int def ref_factory(ref: str) -> str: return f"http://some-domain.org/path/to/{ref}.json#" assert deserialization_schema(Foo, all_refs=True, ref_factory=ref_factory) == { "$schema": "http://json-schema.org/draft/2020-12/schema#", "$ref": "http://some-domain.org/path/to/Foo.json#", } ``` -------------------------------- ### Base Schema for Resolvers using Docstrings Source: https://wyfo.github.io/apischema/0.19/graphql/data_model_and_resolvers Configure apischema to use docstrings for generating base schemas for methods and parameters, enabling automatic inclusion of descriptions in the GraphQL schema. ```python import inspect from dataclasses import dataclass from typing import Any, Callable import docstring_parser from graphql.utilities import print_schema from apischema import schema, settings from apischema.graphql import graphql_schema, resolver from apischema.schemas import Schema @dataclass class Foo: @resolver def bar(self, arg: str) -> int: """bar method :param arg: arg parameter """ ... def method_base_schema(tp: Any, method: Callable, alias: str) -> Schema | None: return schema(description=docstring_parser.parse(method.__doc__).short_description) def parameter_base_schema( method: Callable, parameter: inspect.Parameter, alias: str ) -> Schema | None: for doc_param in docstring_parser.parse(method.__doc__).params: if doc_param.arg_name == parameter.name: return schema(description=doc_param.description) return None settings.base_schema.method = method_base_schema settings.base_schema.parameter = parameter_base_schema def foo() -> Foo: ... schema_ = graphql_schema(query=[foo]) schema_str = '''\ type Query { foo: Foo! } type Foo { """bar method""" bar( """arg parameter""" arg: String! ): Int! }''' assert print_schema(schema_) == schema_str ``` -------------------------------- ### Defining GraphQL Interfaces Source: https://wyfo.github.io/apischema/0.19/graphql/data_model_and_resolvers Demonstrates how to define GraphQL interfaces using the @interface decorator on dataclasses, allowing object types to implement them through inheritance. ```APIDOC ## GraphQL Interface Definition ### Description Use the @interface decorator to mark a class as a GraphQL interface. Classes inheriting from this interface will automatically implement it in the generated schema. ### Request Example ```python from dataclasses import dataclass from apischema.graphql import interface @interface @dataclass class Bar: bar: int @dataclass class Foo(Bar): baz: str ``` ``` -------------------------------- ### Relay Connection Implementation Source: https://wyfo.github.io/apischema/0.19/graphql/relay Defines how to use relay.Connection and relay.Edge to create paginated GraphQL fields. ```APIDOC ## Relay Connection ### Description Provides a generic structure for Relay-compliant connections, including edges and page information. ### Parameters #### Request Body (Resolver Arguments) - **first** (int) - Optional - The number of items to return. - **after** (Cursor) - Optional - The cursor to start pagination after. ### Response #### Success Response (200) - **edges** (Sequence) - A list of edges containing nodes and cursors. - **pageInfo** (PageInfo) - Metadata about the current page, including hasNextPage, hasPreviousPage, startCursor, and endCursor. ``` -------------------------------- ### Customizing Connections and Edges Source: https://wyfo.github.io/apischema/0.19/graphql/relay Explains how to extend the base relay.Connection and relay.Edge classes to include custom fields in the GraphQL schema. ```APIDOC ## Custom Connection/Edge ### Description Subclass relay.Connection or relay.Edge to add custom fields to the connection or edge types in the generated GraphQL schema. ### Implementation - Subclass `relay.Connection` and add fields as dataclass attributes. - Subclass `relay.Edge` and add fields as dataclass attributes. - Pass the custom edge class as a type argument to the generic connection. ``` -------------------------------- ### Implement SQLAlchemy model support Source: https://wyfo.github.io/apischema/0.19/examples/sqlalchemy_support Uses a declarative base class to automatically map SQLAlchemy columns to apischema object fields. Requires defining a base class with __init_subclass__ to register fields. ```python from collections.abc import Collection from inspect import getmembers from itertools import starmap from typing import Any from graphql import print_schema from sqlalchemy import Column, Integer, String from sqlalchemy.ext.declarative import as_declarative from apischema import Undefined, deserialize, serialize from apischema.graphql import graphql_schema from apischema.json_schema import deserialization_schema from apischema.objects import ObjectField, set_object_fields def column_field(name: str, column: Column) -> ObjectField: required = False default: Any = ... if column.default is not None: default = column.default elif column.server_default is not None: default = Undefined elif column.nullable: default = None else: required = True col_type = column.type.python_type if column.nullable: col_type = col_type | None return ObjectField(column.name or name, col_type, required, default=default) # Very basic SQLAlchemy support @as_declarative() class Base: def __init_subclass__(cls, **kwargs): super().__init_subclass__(**kwargs) columns = getmembers(cls, lambda m: isinstance(m, Column)) if not columns: return set_object_fields(cls, starmap(column_field, columns)) class Foo(Base): __tablename__ = "foo" bar = Column(Integer, primary_key=True) baz = Column(String) foo = deserialize(Foo, {"bar": 0}) assert isinstance(foo, Foo) assert foo.bar == 0 assert serialize(Foo, foo) == {"bar": 0, "baz": None} assert deserialization_schema(Foo) == { "$schema": "http://json-schema.org/draft/2020-12/schema#", "type": "object", "properties": { "bar": {"type": "integer"}, "baz": {"type": ["string", "null"], "default": None}, }, "required": ["bar"], "additionalProperties": False, } def foos() -> Collection[Foo] | None: ... schema = graphql_schema(query=[foos]) schema_str = """\ type Query { foos: [Foo!] } type Foo { bar: Int! baz: String }""" assert print_schema(schema) == schema_str ``` -------------------------------- ### Custom Type Schema with apischema Source: https://wyfo.github.io/apischema/0.19/difference_with_pydantic Shows the simpler approach in apischema for defining custom types like RGB. It leverages standard Python types and requires minimal code for schema generation. ```python import re from typing import NamedTuple, NewType import pydantic.validators import apischema # Serialization can only be customized into the enclosing models class RGB(NamedTuple): red: int green: int blue: int # NewType can be used to add schema to conversion source/target ``` -------------------------------- ### Implement Tagged Unions in GraphQL Source: https://wyfo.github.io/apischema/0.19/graphql/data_model_and_resolvers Demonstrates tagged union support in GraphQL via nullable input object types, where multiple tags trigger an error. ```python from dataclasses import dataclass from graphql import graphql_sync from graphql.utilities import print_schema from apischema.graphql import graphql_schema from apischema.tagged_unions import Tagged, TaggedUnion @dataclass class Bar: field: str class Foo(TaggedUnion): bar: Tagged[Bar] baz: Tagged[int] def query(foo: Foo) -> Foo: return foo schema = graphql_schema(query=[query]) schema_str = """\ type Query { query(foo: FooInput!): Foo! } type Foo { bar: Bar baz: Int } type Bar { field: String! } input FooInput { bar: BarInput baz: Int } input BarInput { field: String! }""" assert print_schema(schema) == schema_str query_str = """ { query(foo: {bar: {field: "value"}}) { bar { field } baz } }""" assert graphql_sync(schema, query_str).data == { "query": {"bar": {"field": "value"}, "baz": None} } ``` -------------------------------- ### Automatic Dependency Management in Apischema Source: https://wyfo.github.io/apischema/0.19/validation Demonstrates how Apischema automatically manages validator dependencies. Validators are only executed if all their dependencies are valid, preventing errors from ill-formed fields. ```python from dataclasses import dataclass import pytest from apischema import ValidationError, deserialize, validator @dataclass class PasswordForm: password: str confirmation: str @validator def password_match(self): if self.password != self.confirmation: raise ValueError("password doesn't match its confirmation") with pytest.raises(ValidationError) as err: deserialize(PasswordForm, {"password": "p455w0rd"}) assert err.value.errors == [ # validator is not executed because confirmation is missing {"loc": ["confirmation"], "err": "missing property"} ] ``` -------------------------------- ### Apply Liskov substitution principle Source: https://wyfo.github.io/apischema/0.19/conversions Illustrates how dynamic conversions respect subclass/superclass relationships during serialization and deserialization. ```python from dataclasses import dataclass from apischema import deserialize, serialize @dataclass class Foo: field: int @dataclass class Bar(Foo): other: str def foo_to_int(foo: Foo) -> int: return foo.field def bar_from_int(i: int) -> Bar: return Bar(i, str(i)) assert serialize(Bar, Bar(0, ""), conversion=foo_to_int) == 0 assert deserialize(Foo, 0, conversion=bar_from_int) == Bar(0, "0") ```