aiodbcore

DBCore

ORM that does not require the developer to create models specifically for it. dbcore works with dataclasses and allows you to connect from one interface to both local databases (sqlite+aiosqlite) and remote databases (postgres+asyncpg).

Small example

import os
import asyncio
from dataclasses import dataclass

from aiodbcore import AsyncDBCore


# a model in which only the id field gives out DB membership
@dataclass
class MyModel:
    id: int | None = None
    foo: int = 0
    bar: str = ""


class MyDB(AsyncDBCore[MyModel]):
    # there i can implement my queries
    async def my_simple_query(self) -> list[MyModel]:
        return await self.fetchall(MyModel, where=MyModel.foo > 10)


async def main():
    MyDB.init(os.environ["DB"])  # init db at start of program
    db = MyDB()
    data = await db.my_simple_query()
    first = data[0]
    first.foo += 100
    await db.save(first)
    await db.close_connections()


if __name__ == '__main__':
    asyncio.run(main())

The declaration of the model from the example above is certainly simple and elegant, but this method does not allow the IDE to show type hints, and static typers will complain. Therefore, there is another way to declare models, it will require minimal changes in the code.

from dataclasses import dataclas
from aiodbcore.models import Field

@dataclass
class MyModel:
    # Here you can wrap default values in `Field`
    # if you don't want complaints from static typers,
    # but it is not necessary
    id: Field[int | None] = Field(None)
    foo: Field[int] = 0
    bar: Filed[str] = ""

Sync version

SyncDBCore has the same interface as the async version. You just need to remove all the async and await statements.

class MyDB(SyncDBCore[MyModel]):  # the same model `MyModel` is used
    def my_simple_query(self) -> list[MyModel]:
        return self.fetchall(MyModel, where=MyModel.foo > 10)

Currently only sqlite+sqlite3 is supported.

Context manager declaration

Using the Database class allows you to select synchronous and asynchronous connections using the context manager.

from aiodbcore import Database

class MyDB(Database[MyModel]): ...

MyDB.init("sqlite://db.sqlite")  # In this case, you cannot explicitly specify the library in the database path.

def sync_environment():
    with MyDB() as db:
        data = db.fetchall(MyModel)
        
async def async_environment():
    async with MyDB() as db:
        data = await db.fetchall(MyModel)

Installation

You can install aiodbcore using pip:

pip install aiodbcore

for install all async providers (aiosqlite and asyncpg):

pip install aiodbcore[async]