Usage

Asyncpg-Simpleorm requires python >= 3.6.

To use asyncpg-simpleorm in a project:

import asyncpg_simpleorm

The following example can be ran with docker-compose.

$ git clone https://github.com/m-housh/asyncpg-simpleorm.git
$ docker-compose build orm
$ docker-compose run --rm orm python examples/user.py

Create a base class that can be used for multiple models to inherit from.

db.py

import os
import uuid

import asyncpg
from asyncpg_simpleorm import AsyncModel, Column, PoolManager


DB_USERNAME = os.environ.get('DB_USERNAME', 'postgres')
DB_PASSWORD = os.environ.get('DB_PASSWORD', 'password')
DB_HOST = os.environ.get('DB_HOST', 'localhost')
DB_PORT = os.environ.get('DB_PORT', '5432')
DB_NAME = os.environ.get('DB_NAME', 'postgres')


# We use a ``PoolManager`` here, because all classes will
# share the same manager class.
manager = PoolManager(
    user=DB_USERNAME,
    password=DB_PASSWORD,
    host=DB_HOST,
    port=DB_PORT,
    database=DB_NAME
)


class DbModel(AsyncModel, connection=manager):
    """Base database model class that we inherit from.

    Adds the ``id`` column to all models that inherit from this class.  The
    ``id`` in the database will be found at a column named ``_id``.

    """
    id = Column('_id', default=uuid.uuid4, primary_key=True)

Create a simple user class.

user.py

#!/usr/bin/env python

from examples.db import DbModel
from asyncpg_simpleorm import Column


async def delete_users_table(conn) -> None:
    await conn.execute('DROP TABLE IF EXISTS users')


async def create_users_table(conn) -> None:
    await conn.execute('''
        CREATE TABLE IF NOT EXISTS users (
            _id uuid PRIMARY KEY NOT NULL,
            name text NOT NULL,
            email text
        )
    ''')


class User(DbModel):
    """A simple user class.

    Table Columns
    -------------

    _id : uuid  Primary Key
    name : text  Not Null
    email : text

    """
    # Set the database tablename.  If not supplied then it defaults
    # to lower case version of class name.
    __tablename__ = 'users'

    name = Column()
    email = Column()


async def create_table():
    async with User.connection() as conn:
        await delete_users_table(conn)
        await create_users_table(conn)


async def create_some_users():

    for name in ['foo', 'bar', 'baz']:
        u = User(name=name, email=f'{name}@example.com')
        print(f'Saving user: {u}')
        await u.save()


async def get_users_as_records():
    users = await User.get()

    for u in users:
        print(u)


async def get_users_as_instances():
    users = await User.get(records=False)

    for u in users:
        print(u)


async def get_foo_user():
    print(await User.get_one(name='foo'))


async def delete_all_users():
    for u in await User.get(records=False):
        print(f'Deleting user: {u}')
        await u.delete()


async def main():

    await create_table()

    print("\n\nLet's create some users...")
    await create_some_users()

    print('\n\nGetting users as asyncpg.Records...')
    await get_users_as_records()


    print('\n\nGetting users as User instances...')
    await get_users_as_instances()

    print("\n\nGetting 'foo' user")
    await get_foo_user()

    print('\n\nDeleting users...')
    await delete_all_users()

    print('\n\nDropping users table...')
    async with User.connection() as conn:
        await delete_users_table(conn)


if __name__ == '__main__':

    import asyncio

    asyncio.get_event_loop().run_until_complete(main())

Running this example should output something like the following.

output

Let's create some users...
Saving user: User(name='foo', email='foo@example.com', id=0cc2385f-d855-4d55-a7cc-398b176e120f)
Saving user: User(name='bar', email='bar@example.com', id=cebf4997-f857-49f1-9391-4835003eb980)
Saving user: User(name='baz', email='baz@example.com', id=c722d54e-5b68-4ea5-85c9-d8e31ba95bfa)


Getting users as asyncpg.Records...
<Record name='foo' email='foo@example.com' _id=UUID('0cc2385f-d855-4d55-a7cc-398b176e120f')>
<Record name='bar' email='bar@example.com' _id=UUID('cebf4997-f857-49f1-9391-4835003eb980')>
<Record name='baz' email='baz@example.com' _id=UUID('c722d54e-5b68-4ea5-85c9-d8e31ba95bfa')>


Getting users as User instances...
User(name='foo', email='foo@example.com', id=0cc2385f-d855-4d55-a7cc-398b176e120f)
User(name='bar', email='bar@example.com', id=cebf4997-f857-49f1-9391-4835003eb980)
User(name='baz', email='baz@example.com',
id=c722d54e-5b68-4ea5-85c9-d8e31ba95bfa)


Getting 'foo' user
<Record name='foo' email='foo@example.com' _id=UUID('0cc2385f-d855-4d55-a7cc-398b176e120f')>


Deleting users...
Deleting user: User(name='foo', email='foo@example.com', id=0cc2385f-d855-4d55-a7cc-398b176e120f)
Deleting user: User(name='bar', email='bar@example.com', id=cebf4997-f857-49f1-9391-4835003eb980)
Deleting user: User(name='baz', email='baz@example.com', id=c722d54e-5b68-4ea5-85c9-d8e31ba95bfa)


Dropping users table...