Skip to content

Ormdantic

ormdantic.orm.Ormdantic 

Ormdantic(connection)

Ormdantic provides a way to create ORM models and schemas.

Register models as ORM models and create schemas

PARAMETER DESCRIPTION
connection

TYPE: str

Source code in ormdantic/orm.py 
26
27
28
29
30
31
def __init__(self, connection: str) -> None:
    """Register models as ORM models and create schemas"""
    self._metadata: MetaData | None = None
    self._crud_generators: dict[Type, CRUD] = {}  # type: ignore
    self._engine = create_async_engine(connection)
    self._table_map: Map = Map()

table 

table(tablename=None, *, pk, indexed=None, unique=None, unique_constraints=None, back_references=None)

Register a model as a database table.

PARAMETER DESCRIPTION
tablename

TYPE: str | None DEFAULT: None

pk

TYPE: str

indexed

TYPE: list[str] | None DEFAULT: None

unique

TYPE: list[str] | None DEFAULT: None

unique_constraints

TYPE: list[list[str]] | None DEFAULT: None

back_references

TYPE: dict[str, str] | None DEFAULT: None

Source code in ormdantic/orm.py 
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
def table(
    self,
    tablename: str | None = None,
    *,
    pk: str,
    indexed: list[str] | None = None,
    unique: list[str] | None = None,
    unique_constraints: list[list[str]] | None = None,
    back_references: dict[str, str] | None = None,
) -> Callable[[Type[ModelType]], Type[ModelType]]:
    """Register a model as a database table."""

    def _wrapper(cls: Type[ModelType]) -> Type[ModelType]:
        """Decorator function."""
        tablename_ = tablename or snake_case(cls.__name__)
        cls_back_references = back_references or {}
        table_metadata = OrmTable[ModelType](
            model=cls,
            tablename=tablename_,
            pk=pk,
            indexed=indexed or [],
            unique=unique or [],
            unique_constraints=unique_constraints or [],
            columns=[
                field
                for field in cls.__fields__
                if field not in cls_back_references
            ],
            relationships={},
            back_references=cls_back_references,
        )
        self._table_map.model_to_data[cls] = table_metadata
        self._table_map.name_to_data[tablename_] = table_metadata
        return cls

    return _wrapper

init async 

init()

Initialize ORM models.

Source code in ormdantic/orm.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
async def init(self) -> None:
    """Initialize ORM models."""
    # Populate relation information.
    for table_data in self._table_map.name_to_data.values():
        rels = self.get(table_data)
        table_data.relationships = rels
    # Now that relation information is populated generate tables.
    self._metadata = MetaData()
    for table_data in self._table_map.name_to_data.values():
        self._crud_generators[table_data.model] = CRUD(
            table_data,
            self._table_map,
            self._engine,
        )
    await Table(self._engine, self._metadata, self._table_map).init()
    async with self._engine.begin() as conn:
        await conn.run_sync(self._metadata.create_all)

get 

get(table_data)

Get relationships for a given table.

PARAMETER DESCRIPTION
table_data

TYPE: OrmTable[ModelType]

Source code in ormdantic/orm.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def get(self, table_data: OrmTable[ModelType]) -> dict[str, Relationship]:
    """Get relationships for a given table."""
    relationships = {}
    for field_name, field in table_data.model.__fields__.items():
        related_table = self._get_related_table(field)
        if related_table is None:
            continue
        if back_reference := table_data.back_references.get(field_name):
            relationships[field_name] = self._get_many_relationship(
                field_name, back_reference, table_data, related_table
            )

            continue
        if get_origin(field.outer_type_) == list or field.type_ == ForwardRef(
            f"{related_table.model.__name__}"
        ):
            raise UndefinedBackReferenceError(
                table_data.tablename, related_table.tablename, field_name
            )

        args = get_args(field.type_)
        correct_type = (
            related_table.model.__fields__[related_table.pk].type_ in args
        )
        origin = get_origin(field.type_)
        if not args or origin != UnionType or not correct_type:
            raise MustUnionForeignKeyError(
                table_data.tablename,
                related_table.tablename,
                field_name,
                related_table.model,
                related_table.model.__fields__[related_table.pk].type_.__name__,
            )

        relationships[field_name] = Relationship(
            foreign_table=related_table.tablename
        )

    return relationships