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.

Source Distribution

asyncpg-utils-0.3.0.tar.gz (6.4 kB view details)

Uploaded Source

Built Distribution

asyncpg_utils-0.3.0-py3-none-any.whl (7.9 kB view details)

Uploaded Python 3

File details

Details for the file asyncpg-utils-0.3.0.tar.gz.

File metadata

File hashes

Hashes for asyncpg-utils-0.3.0.tar.gz
Algorithm Hash digest
SHA256 e7864255b43d7ba782dff54495af085959ae855b8671f539a4044612b8617a47
MD5 b3230b7b7a81a18427b2687d4107de76
BLAKE2b-256 a4f2fb2a80a0d375acbea7e591e670d17599cfcc19990207d87bb4d8e7d780c4

See more details on using hashes here.

File details

Details for the file asyncpg_utils-0.3.0-py3-none-any.whl.

File metadata

File hashes

Hashes for asyncpg_utils-0.3.0-py3-none-any.whl
Algorithm Hash digest
SHA256 047caa3a8d66ddb1b401eb1ee12ea0c69a7f472d0e2d6c60d5a267b2615b0a29
MD5 5bd834fa3760e0aeabf9679d108b8cec
BLAKE2b-256 2ca3c57a69e76b4e970e2ccac7e58706018b96d8d5a348ee6703fbfff373def3

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page