apischema

https://wyfo.github.io/apischema/
Admin
apischema is a Python library for JSON and GraphQL (de)serialization, JSON schema generation, and...
Tokens:47,261
Snippets:378
Trust Score:9.7
Update:6 months ago
Show doc for...
Tokens:
Raw
### 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")
```
Tokens:
Raw