Danio

Danio is a ORM for python asyncio world.It is designed to make getting easy and clearly.It builds on python's dataclass and encode's databases

Features

• keep OOM in mind, custom your Field and Model behavior easily
• type hints any where, no more need to memorize words your field names any more
• base CRUD operation, transactions, lock and so on
• signals like before save, after save and so on
• complex operation like bulk create, upsert, create or update and so on
• assist model schema migration
• support MySQL/PostgreSQL/SQLite
• hints generation

install

pip install danio

Documents

Danio Document

Glance

db = danio.Database(
    "mysql://root:letmein@server:3306/test",
    maxsize=3,
    charset="utf8mb4",
    use_unicode=True,
    connect_timeout=60,
)

@dataclasses.dataclass
class User(danio.Model):
    # auto generated by danio:
    # --------------------Danio Hints--------------------
    # TABLE NAME: user
    # TABLE IS MIGRATED!
    ID: typing.ClassVar[danio.Field]  # "id" serial PRIMARY KEY NOT NULL
    NAME: typing.ClassVar[danio.Field]  # "name" varchar(255)  NOT NULL
    AGE: typing.ClassVar[danio.Field]  # "age" int  NOT NULL
    CREATED_AT: typing.ClassVar[
        danio.Field
    ]  # "created_at" timestamp without time zone  NOT NULL
    UPDATED_AT: typing.ClassVar[
        danio.Field
    ]  # "updated_at" timestamp without time zone  NOT NULL
    GENDER: typing.ClassVar[danio.Field]  # "gender" int  NOT NULL
    # --------------------Danio Hints--------------------

    class Gender(enum.Enum):
        MALE = 0
        FEMALE = 1
        OTHER = 2

    id: typing.Annotated[int, danio.IntField(primary=True, type="serial")] = 0
    name: typing.Annotated[str, danio.CharField(comment="User name")] = ""
    age: typing.Annotated[int, danio.IntField] = 0
    created_at: typing.Annotated[
        datetime.datetime,
        danio.DateTimeField(type="timestamp without time zone", comment="when created"),
    ] = dataclasses.field(default_factory=datetime.datetime.now)
    updated_at: typing.Annotated[
        datetime.datetime,
        danio.DateTimeField(type="timestamp without time zone", comment="when updated"),
    ] = dataclasses.field(default_factory=datetime.datetime.now)
    gender: typing.Annotated[Gender, danio.IntField(enum=Gender)] = Gender.MALE


    async def before_create(self, validate=True):
        await super().before_create(validate=True)

    async def before_update(self, validate=True):
        self.updated_at = datetime.datetime.now()
        await super().before_update(validate=True)

    async def validate(self):
        await super().validate()
        if not self.name:
            raise danio.ValidateException("Empty name!")

    @classmethod
    def get_database(
        cls, operation: danio.Operation, table: str, *args, **kwargs
    ) -> danio.Database:
        return db

# base CRUD
user = await User(name="batman").save()
user = await User.where(User.NAME == "batman").fetch_one()
user.gender = User.Gender.MALE
await user.save()
await user.delete()
# sql chain
await User.where(User.NAME != "").limit(10).fetch_all()
# multi where condition
await User.where(User.ID != 1, User.NAME != "").fetch_all()
await User.where(User.ID != 1).where(User.NAME != "").fetch_all()
await User.where(User.ID <= 10, User.ID >= 20, is_and=False).fetch_all()
# complicated expression
await User.where(User.ID == 1).update(age=(User.AGE + 1) / (User.AGE / 12) - 2)
await User.where((User.AGE + 1) == 3).fetch_all()
# complicated sql operation
await User.where(User.ID == u.id).update(
    age=User.AGE.case(User.AGE > 10, 1, default=18).case(User.AGE <= 0, 10)
)
created, updated = await UserProfile.upsert(
    [
        dict(id=1, name="upsert"),
    ],
    update_fields=["name"],
)
# bulk operation
await User.bulk_create([User(name=f"user_{i}") for i in range(10)])
await User.bulk_update(await User.fetch_all())
await User.bulk_delete(await User.fetch_all())
# shortcut
user, created = await User(id=1, name="created?").get_or_create(
    key_fields=(User.ID,)
)
user, created, updated = await User(id=2, name="updated?").create_or_update(
    key_fields=(User.ID,)
)