Skip to main content

Utilities for Asyncpg

Project description

TravisCI Build Status Coverage Status Requirements Status Version


Utilities for Asyncpg.

How to install

pip install asyncpg-utils

How to Use

Database

# database.py
import asyncio
from datetime import date

from asyncpg_utils.databases import Database

loop = asyncio.get_event_loop()
database = Database('postgresql://postgres:postgres@localhost/asyncpg-utils')


async def create_table():
    conn = await database.get_connection()
    await conn.execute(
        """
        CREATE TABLE IF NOT EXISTS users(
            id serial PRIMARY KEY,
            name text,
            dob date
        )
        """
    )
    await conn.close()
    return True


async def insert_row(data):
    return await database.insert('users', data)


async def query_all():
    return await database.query(
        """
        SELECT * FROM users
        """
    )


async def query_one():
    return await database.query_one(
        """
        SELECT * FROM users
        WHERE name = $1
        """,
        'Jane Doe'
    )


async def main():
    print('create_table users, {!r}'.format(await create_table()))
    print('insert row, {!r}'.format(await insert_row({'name': 'John Doe', 'dob': date(2000, 1, 1)})))
    print('insert row, {!r}'.format(await insert_row({'name': 'Jane Doe', 'dob': date(2000, 1, 1)})))
    print('query all results, {!r}'.format(await query_all()))
    print('query one result, {!r}'.format(await query_one()))

loop.run_until_complete(main())
# python database.py
create_table users, True
insert row, <Record id=1 name='John Doe' dob=datetime.date(2000, 1, 1)>
insert row, <Record id=2 name='Jane Doe' dob=datetime.date(2000, 1, 1)>
query all results, [<Record id=1 name='John Doe' dob=datetime.date(2000, 1, 1)>, <Record id=2 name='Jane Doe' dob=datetime.date(2000, 1, 1)>]
query one result, <Record id=2 name='Jane Doe' dob=datetime.date(2000, 1, 1)>

PoolDatabase

# database_pool.py
import asyncio
from datetime import date

from asyncpg_utils.databases import PoolDatabase

loop = asyncio.get_event_loop()
database = PoolDatabase('postgresql://postgres:postgres@localhost/asyncpg-utils')


async def create_table():
    conn = await database.get_connection()
    await conn.execute(
        """
        CREATE TABLE IF NOT EXISTS users(
            id serial PRIMARY KEY,
            name text,
            dob date
        )
        """
    )
    await conn.close()
    return True


async def insert_row(data):
    return await database.insert('users', data)


async def query_all():
    return await database.query(
        """
        SELECT * FROM users
        """
    )


async def query_one():
    return await database.query_one(
        """
        SELECT * FROM users
        WHERE name = $1
        """,
        'Jane Doe'
    )


async def main():
    print('create_table users, {!r}'.format(await create_table()))
    print('insert row, {!r}'.format(await insert_row({'name': 'John Doe', 'dob': date(2000, 1, 1)})))
    print('insert row, {!r}'.format(await insert_row({'name': 'Jane Doe', 'dob': date(2000, 1, 1)})))
    print('query all results, {!r}'.format(await query_all()))
    print('query one result, {!r}'.format(await query_one()))

loop.run_until_complete(main())
# python database_pool.py
create_table users, True
insert row, <Record id=1 name='John Doe' dob=datetime.date(2000, 1, 1)>
insert row, <Record id=2 name='Jane Doe' dob=datetime.date(2000, 1, 1)>
query all results, [<Record id=1 name='John Doe' dob=datetime.date(2000, 1, 1)>, <Record id=2 name='Jane Doe' dob=datetime.date(2000, 1, 1)>]
query one result, <Record id=2 name='Jane Doe' dob=datetime.date(2000, 1, 1)>

TableManager

# table_manager.py
import asyncio
from datetime import date

from asyncpg_utils.databases import Database
from asyncpg_utils.managers import TableManager

loop = asyncio.get_event_loop()
database = Database('postgresql://postgres:postgres@localhost/asyncpg-utils')
table_manager = TableManager(database, 'users', pk_field='id', hooks=None)
user_data = {
    'name': 'Allisson',
    'dob': date(1983, 2, 9)
}


async def create_table():
    conn = await database.get_connection()
    await conn.execute(
        """
        CREATE TABLE IF NOT EXISTS users(
            id serial PRIMARY KEY,
            name text,
            dob date
        )
        """
    )
    await conn.close()
    return True


async def table_manager_create():
    print('table_manager.create, row={!r}'.format(await table_manager.create(user_data)))


async def table_manager_list():
    print('table_manager.list, rows={!r}'.format(await table_manager.list()))
    print('table_manager.list, only_name_field, rows={!r}'.format(await table_manager.list(fields=['name'])))
    print('table_manager.list, filter_by_id, rows={!r}'.format(await table_manager.list(filters={'id': 999999})))


async def table_manager_detail():
    print('table_manager.detail, row={!r}'.format(await table_manager.detail(1)))
    print('table_manager.detail, only_name_field, row={!r}'.format(await table_manager.detail(1, fields=['name'])))


async def table_manager_update():
    user_data['name'] = 'John Doe'
    print('table_manager.update, row={!r}'.format(await table_manager.update(1, user_data)))


async def table_manager_delete():
    print('table_manager.delete, result={!r}'.format(await table_manager.delete(1)))


async def main():
    print('create_table users, {!r}'.format(await create_table()))
    await table_manager_create()
    await table_manager_list()
    await table_manager_detail()
    await table_manager_update()
    await table_manager_delete()

loop.run_until_complete(main())
# python table_manager.py
create_table users, True
table_manager.create, row=<Record id=1 name='Allisson' dob=datetime.date(1983, 2, 9)>
table_manager.list, rows=[<Record id=1 name='Allisson' dob=datetime.date(1983, 2, 9)>]
table_manager.list, only_name_field, rows=[<Record name='Allisson'>]
table_manager.list, filter_by_id, rows=[]
table_manager.detail, row=<Record id=1 name='Allisson' dob=datetime.date(1983, 2, 9)>
table_manager.detail, only_name_field, row=<Record name='Allisson'>
table_manager.update, row=<Record id=1 name='John Doe' dob=datetime.date(1983, 2, 9)>
table_manager.delete, result=True

Table Manager Hook

# table_manager_hook.py
import asyncio
from datetime import date

from asyncpg_utils.databases import Database
from asyncpg_utils.managers import AbstractHook, TableManager

loop = asyncio.get_event_loop()
database = Database('postgresql://postgres:postgres@localhost/asyncpg-utils')
user_data = {
    'name': 'Allisson',
    'dob': date(1983, 2, 9)
}


class TestHook(AbstractHook):
    async def pre_create(self, data):
        print('pre_create, data={!r}'.format(data))

    async def post_create(self, row):
        print('post_create, row={!r}'.format(row))

    async def pre_list(self, fields, filters):
        print('pre_list, fields={!r}, filters={!r}'.format(fields, filters))

    async def post_list(self, rows):
        print('post_list, rows={!r}'.format(rows))

    async def pre_detail(self, pk, fields):
        print('pre_detail, pk={!r}, fields={!r}'.format(pk, fields))

    async def post_detail(self, row):
        print('post_detail, row={!r}'.format(row))

    async def pre_update(self, pk, data):
        print('pre_update, pk={!r}, data={!r}'.format(pk, data))

    async def post_update(self, row):
        print('post_update, row={!r}'.format(row))

    async def pre_delete(self, pk):
        print('pre_delete, pk={!r}'.format(pk))

    async def post_delete(self, pk):
        print('post_delete, pk={!r}'.format(pk))


table_manager = TableManager(database, 'users', pk_field='id', hooks=(TestHook,))


async def create_table():
    conn = await database.get_connection()
    await conn.execute(
        """
        CREATE TABLE IF NOT EXISTS users(
            id serial PRIMARY KEY,
            name text,
            dob date
        )
        """
    )
    await conn.close()
    return True


async def main():
    print('create_table users, {!r}'.format(await create_table()))
    await table_manager.create(user_data)
    await table_manager.list()
    await table_manager.detail(1)
    user_data['name'] = 'John Doe'
    await table_manager.update(1, user_data)
    await table_manager.delete(1)


loop.run_until_complete(main())
# python table_manager_hook.py
create_table users, True
pre_create, data={'name': 'Allisson', 'dob': datetime.date(1983, 2, 9)}
post_create, row=<Record id=2 name='Allisson' dob=datetime.date(1983, 2, 9)>
pre_list, fields=None, filters={}
post_list, rows=[<Record id=2 name='Allisson' dob=datetime.date(1983, 2, 9)>]
pre_detail, pk=1, fields=None
post_detail, row=None
pre_update, pk=1, data={'name': 'John Doe', 'dob': datetime.date(1983, 2, 9)}
post_update, row=None
pre_delete, pk=1
post_delete, pk=1

Check https://github.com/allisson/asyncpg-utils/tree/master/examples for more code examples.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for asyncpg-utils, version 0.3.0
Filename, size File type Python version Upload date Hashes
Filename, size asyncpg_utils-0.3.0-py3-none-any.whl (7.9 kB) File type Wheel Python version py3 Upload date Hashes View
Filename, size asyncpg-utils-0.3.0.tar.gz (6.4 kB) File type Source Python version None Upload date Hashes View

Supported by

Pingdom Pingdom Monitoring Google Google Object Storage and Download Analytics Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN DigiCert DigiCert EV certificate StatusPage StatusPage Status page