ORM

This module includes the asynchronous ORM functionalities for the SQLite3 subpackage.

Examples

Field and Model Definition

Define fields and models using the ORM functionalities.

import asyncio
from DbUnify.SQLite3.aio.ORM import ORMManager, Model, Field
from DbUnify.SQLite3.data import *

class User(Model):
    id = Field(DataType.INTEGER, constraints=[Rules.PRIMARY_KEY, Rules.AUTOINCREMENT])
    name = Field(DataType.TEXT, constraints=[Rules.NOT_NULL])
    email = Field(DataType.TEXT, constraints=[Rules.UNIQUE])

async def main():
    orm_manager = ORMManager(db_name='example.db')
    await orm_manager.connect()
    User.set_manager(orm_manager)
    await User.create_table()
    await orm_manager.close()

asyncio.run(main())

Fetch All Rows

Fetch all rows from a table.

import asyncio
from DbUnify.SQLite3.aio.ORM import ORMManager, Model, Field
from DbUnify.SQLite3.data import *

class User(Model):
    id = Field(DataType.INTEGER, constraints=[Rules.PRIMARY_KEY, Rules.AUTOINCREMENT])
    name = Field(DataType.TEXT, constraints=[Rules.NOT_NULL])
    email = Field(DataType.TEXT, constraints=[Rules.UNIQUE])

async def main():
    async with ORMManager(db_name='example.db') as orm_manager:
        User.set_manager(orm_manager)
        await User.create_table()
        await User.insert_row(values={'name': 'John Doe', 'email': 'john@example.com'})
        users = await User.fetch_all(query='SELECT * FROM user')
        print("All users:", users)

asyncio.run(main())

Create Table

Create a table using the ORM functionalities.

import asyncio
from DbUnify.SQLite3.aio.ORM import ORMManager, Model, Field
from DbUnify.SQLite3.data import *

class User(Model):
    id = Field(DataType.INTEGER, constraints=[Rules.PRIMARY_KEY, Rules.AUTOINCREMENT])
    name = Field(DataType.TEXT, constraints=[Rules.NOT_NULL])
    email = Field(DataType.TEXT, constraints=[Rules.UNIQUE])

async def main():
    orm_manager = ORMManager(db_name='example.db')
    await orm_manager.connect()
    User.set_manager(orm_manager)
    await User.create_table()
    await orm_manager.close()

asyncio.run(main())

Insert Row

Insert a row into a table.

import asyncio
from DbUnify.SQLite3.aio.ORM import ORMManager, Model, Field
from DbUnify.SQLite3.data import *

class User(Model):
    id = Field(DataType.INTEGER, constraints=[Rules.PRIMARY_KEY, Rules.AUTOINCREMENT])
    name = Field(DataType.TEXT, constraints=[Rules.NOT_NULL])
    email = Field(DataType.TEXT, constraints=[Rules.UNIQUE])

async def main():
    orm_manager = ORMManager(db_name='example.db')
    await orm_manager.connect()
    User.set_manager(orm_manager)
    await User.create_table()
    await User.insert_row(values={'name': 'John Doe', 'email': 'john@example.com'})
    await orm_manager.close()

asyncio.run(main())

Select Rows

Select rows from a table.

import asyncio
from DbUnify.SQLite3.aio.ORM import ORMManager, Model, Field
from DbUnify.SQLite3.data import *

class User(Model):
    id = Field(DataType.INTEGER, constraints=[Rules.PRIMARY_KEY, Rules.AUTOINCREMENT])
    name = Field(DataType.TEXT, constraints=[Rules.NOT_NULL])
    email = Field(DataType.TEXT, constraints=[Rules.UNIQUE])

async def main():
    orm_manager = ORMManager(db_name='example.db')
    await orm_manager.connect()
    User.set_manager(orm_manager)
    await User.create_table()
    await User.insert_row(values={'name': 'John Doe', 'email': 'john@example.com'})
    users = await User.select()
    print("Users:", users)
    await orm_manager.close()

asyncio.run(main())

Update Row

Update a row in the table.

import asyncio
from DbUnify.SQLite3.aio.ORM import ORMManager, Model, Field
from DbUnify.SQLite3.data import *

class User(Model):
    id = Field(DataType.INTEGER, constraints=[Rules.PRIMARY_KEY, Rules.AUTOINCREMENT])
    name = Field(DataType.TEXT, constraints=[Rules.NOT_NULL])
    email = Field(DataType.TEXT, constraints=[Rules.UNIQUE])

async def main():
    orm_manager = ORMManager(db_name='example.db')
    await orm_manager.connect()
    User.set_manager(orm_manager)
    await User.create_table()
    await User.insert_row(values={'name': 'John Doe', 'email': 'john@example.com'})
    await User.update_row(values={'name': 'Jane Doe'}, condition='email = ?', 'john@example.com')
    users = await User.select()
    print("Users after update:", users)
    await orm_manager.close()

asyncio.run(main())

Delete Row

Delete a row from the table.

import asyncio
from DbUnify.SQLite3.aio.ORM import ORMManager, Model, Field
from DbUnify.SQLite3.data import *

class User(Model):
    id = Field(DataType.INTEGER, constraints=[Rules.PRIMARY_KEY, Rules.AUTOINCREMENT])
    name = Field(DataType.TEXT, constraints=[Rules.NOT_NULL])
    email = Field(DataType.TEXT, constraints=[Rules.UNIQUE])

async def main():
    orm_manager = ORMManager(db_name='example.db')
    await orm_manager.connect()
    User.set_manager(orm_manager)
    await User.create_table()
    await User.insert_row(values={'name': 'John Doe', 'email': 'john@example.com'})
    await User.delete_row(condition='email = ?', 'john@example.com')
    users = await User.select()
    print("Users after deletion:", users)
    await orm_manager.close()

asyncio.run(main())

Drop Table

Drop a table from the database.

import asyncio
from DbUnify.SQLite3.aio.ORM import ORMManager, Model, Field
from DbUnify.SQLite3.data import *

class User(Model):
    id = Field(DataType.INTEGER, constraints=[Rules.PRIMARY_KEY, Rules.AUTOINCREMENT])
    name = Field(DataType.TEXT, constraints=[Rules.NOT_NULL])
    email = Field(DataType.TEXT, constraints=[Rules.UNIQUE])

async def main():
    orm_manager = ORMManager(db_name='example.db')
    await orm_manager.connect()
    User.set_manager(orm_manager)
    await User.create_table()
    await User.drop_table()
    await orm_manager.close()

asyncio.run(main())

Apply Migrations

Apply migrations to the database schema based on the model.

import asyncio
from DbUnify.SQLite3.aio.ORM import ORMManager, Model, Field
from DbUnify.SQLite3.data import *

class User(Model):
    id = Field(DataType.INTEGER, constraints=[Rules.PRIMARY_KEY, Rules.AUTOINCREMENT])
    name = Field(DataType.TEXT, constraints=[Rules.NOT_NULL])
    email = Field(DataType.TEXT, constraints=[Rules.UNIQUE])

async def main():
    async with ORMManager(db_name='example.db') as orm_manager:
        User.set_manager(orm_manager)
        await User.create_table()
        await orm_manager.apply_migrations(User)

asyncio.run(main())

Map Model

Map a model to a database table and create it if it doesn’t exist.

import asyncio
from DbUnify.SQLite3.aio.ORM import ORMManager, Model, Field
from DbUnify.SQLite3.data import *

class User(Model):
    id = Field(DataType.INTEGER, constraints=[Rules.PRIMARY_KEY, Rules.AUTOINCREMENT])
    name = Field(DataType.TEXT, constraints=[Rules.NOT_NULL])
    email = Field(DataType.TEXT, constraints=[Rules.UNIQUE])

async def main():
    async with ORMManager(db_name='example.db') as orm_manager:
        await orm_manager.map_model(User)

asyncio.run(main())