Utilities for Asyncpg
Project description
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)
Built Distribution
File details
Details for the file asyncpg-utils-0.3.0.tar.gz
.
File metadata
- Download URL: asyncpg-utils-0.3.0.tar.gz
- Upload date:
- Size: 6.4 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
File hashes
Algorithm | Hash digest | |
---|---|---|
SHA256 | e7864255b43d7ba782dff54495af085959ae855b8671f539a4044612b8617a47 |
|
MD5 | b3230b7b7a81a18427b2687d4107de76 |
|
BLAKE2b-256 | a4f2fb2a80a0d375acbea7e591e670d17599cfcc19990207d87bb4d8e7d780c4 |
File details
Details for the file asyncpg_utils-0.3.0-py3-none-any.whl
.
File metadata
- Download URL: asyncpg_utils-0.3.0-py3-none-any.whl
- Upload date:
- Size: 7.9 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
File hashes
Algorithm | Hash digest | |
---|---|---|
SHA256 | 047caa3a8d66ddb1b401eb1ee12ea0c69a7f472d0e2d6c60d5a267b2615b0a29 |
|
MD5 | 5bd834fa3760e0aeabf9679d108b8cec |
|
BLAKE2b-256 | 2ca3c57a69e76b4e970e2ccac7e58706018b96d8d5a348ee6703fbfff373def3 |