-
-
Notifications
You must be signed in to change notification settings - Fork 0
/
sqlalchemy_support.py
80 lines (65 loc) · 2.05 KB
/
sqlalchemy_support.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
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