OrmCrud

ormdantic.generator._crud.OrmCrud

OrmCrud(table_data, table_map, engine)

Bases: Generic[ModelType]

Provides DB CRUD methods and table information for a model.

Initialize OrmCrud.

PARAMETER	DESCRIPTION
table_data	TYPE: OrmTable
table_map	TYPE: Map
engine	TYPE: AsyncEngine

Source code in ormdantic/generator/_crud.py

def __init__(
    self,
    table_data: OrmTable,  # type: ignore
    table_map: Map,
    engine: AsyncEngine,
) -> None:
    """Initialize OrmCrud."""
    self._engine = engine
    self._table_map = table_map
    self._table_data = table_data
    self.tablename = table_data.tablename
    self.columns = table_data.columns

tablename instance-attribute

tablename = tablename

columns instance-attribute

columns = columns

find_one async

find_one(pk, depth=0)

Find a model instance by primary key.

PARAMETER	DESCRIPTION
pk	TYPE: Any
depth	TYPE: int DEFAULT: 0

Source code in ormdantic/generator/_crud.py

async def find_one(self, pk: Any, depth: int = 0) -> ModelType | None:
    """Find a model instance by primary key."""
    result = await self._execute_query(
        OrmField(self._table_data, self._table_map).get_find_one_query(pk, depth)
    )
    return OrmSerializer[ModelType | None](
        table_data=self._table_data,
        table_map=self._table_map,
        result_set=result,
        is_array=False,
        depth=depth,
    ).deserialize()

find_many async

find_many(where=None, order_by=None, order=Order.asc, limit=0, offset=0, depth=0)

Find many model instances.

PARAMETER	DESCRIPTION
where	TYPE: dict[str, Any] | None DEFAULT: None
order_by	TYPE: list[str] | None DEFAULT: None
order	TYPE: Order DEFAULT: asc
limit	TYPE: int DEFAULT: 0
offset	TYPE: int DEFAULT: 0
depth	TYPE: int DEFAULT: 0

Source code in ormdantic/generator/_crud.py

async def find_many(
    self,
    where: dict[str, Any] | None = None,
    order_by: list[str] | None = None,
    order: Order = Order.asc,
    limit: int = 0,
    offset: int = 0,
    depth: int = 0,
) -> Result[ModelType]:
    """Find many model instances."""
    result = await self._execute_query(
        OrmField(self._table_data, self._table_map).get_find_many_query(
            where, order_by, order, limit, offset, depth
        )
    )
    deserialized_data = OrmSerializer[ModelType | None](
        table_data=self._table_data,
        table_map=self._table_map,
        result_set=result,
        is_array=True,
        depth=depth,
    ).deserialize()
    return Result(
        offset=offset,
        limit=limit,
        data=deserialized_data or [],
    )

insert async

insert(model_instance)

Insert a model instance.

PARAMETER	DESCRIPTION
model_instance	TYPE: ModelType

Source code in ormdantic/generator/_crud.py

async def insert(self, model_instance: ModelType) -> ModelType:
    """Insert a model instance."""
    await self._execute_query(
        OrmQuery(model_instance, self._table_map).get_insert_query()
    )
    return model_instance

update async

update(model_instance)

Update a record.

PARAMETER	DESCRIPTION
model_instance	TYPE: ModelType

Source code in ormdantic/generator/_crud.py

async def update(self, model_instance: ModelType) -> ModelType:
    """Update a record."""
    await self._execute_query(
        OrmQuery(model_instance, self._table_map).get_update_queries()
    )
    return model_instance

upsert async

upsert(model_instance)

Insert a record if it does not exist, else update it.

PARAMETER	DESCRIPTION
model_instance	TYPE: ModelType

Source code in ormdantic/generator/_crud.py

async def upsert(self, model_instance: ModelType) -> ModelType:
    """Insert a record if it does not exist, else update it."""

    await self._execute_query(
        OrmQuery(model_instance, self._table_map).get_upsert_query()
    )
    return model_instance

delete async

delete(pk)

Delete a model instance by primary key.

PARAMETER	DESCRIPTION
pk	TYPE: Any

Source code in ormdantic/generator/_crud.py

async def delete(self, pk: Any) -> bool:
    """Delete a model instance by primary key."""
    await self._execute_query(
        OrmField(self._table_data, self._table_map).get_delete_query(pk)
    )
    return True

count async

count(where=None, depth=0)

Count records.

PARAMETER	DESCRIPTION
where	TYPE: dict[str, Any] | None DEFAULT: None
depth	TYPE: int DEFAULT: 0

Source code in ormdantic/generator/_crud.py

async def count(self, where: dict[str, Any] | None = None, depth: int = 0) -> int:
    """Count records."""
    result = await self._execute_query(
        OrmField(self._table_data, self._table_map).get_count_query(where, depth)
    )
    return result.scalar()