Examples¶
Asyncpg-Simpleorm aids in the creation of query statements for use with
asyncpg async postgres connector. This package allows you to create model
classes (similar to sqlalchemy syntax) and can execute common database
operations.
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
db.py¶
The following is just some boiler plate to make a connection manager that is required by our database model(s) we will create.
"""
examples/db.py
--------------
We use a ``PoolManager`` here, but a single connection manager can be
created using orm.ConnectionManager(...)
The ``PoolManager`` mimics the ``asyncpg.create_pool`` method, passing any
*args and/or **kwargs to that method, and the ``ConnectionManager`` mimics
``asyncpg.connect``.
"""
import os
import asyncpg_simpleorm as orm
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')
pool_manager = orm.PoolManager(
user=DB_USERNAME,
password=DB_PASSWORD,
host=DB_HOST,
port=DB_PORT,
database=DB_NAME,
command_timeout=60
)
user.py¶
The primary class exposed by Asyncpg-Simpleorm is the AsyncModel class.
The AsyncModel model class uses __init_subclass__ (new in python 3.6),
to allow key word parameters in the class declaration of all subclasses. We
use this to set a manager on the class, that is used to manage the connection
to the database, that a subclass uses in it’s database interactions.
Here we use a PoolManager instance defined in db.py, we very
well could/should use the ConnectionManager class that just manages a single
connection, however if you are using the same manager for multiple database
models, then the PoolManager makes more sense.
"""
examples/user.py
----------------
A simple user model to show how some features of the ``asyncpg_simpleorm``
package.
"""
import uuid
from examples.db import pool_manager
import asyncpg_simpleorm as orm
class User(orm.AsyncModel, connection=pool_manager):
"""A simple user class.
Table Columns
-------------
_id : uuid Primary Key
name : varchar(40)
email : varchar(100)
"""
# Set the database tablename. If not supplied then it defaults
# to lower case version of class name.
__tablename__ = 'users'
return_records = False
id = orm.Column('_id', orm.UUID(), default=uuid.uuid4, primary_key=True)
name = orm.Column(orm.String(40))
email = orm.Column(orm.String(100))
run.py¶
#!/usr/bin/env python
"""
examples/run.py
---------------
"""
from examples.user import User
import asyncpg_simpleorm as orm
async def create_some_users():
"""Add's some user's to the database.
"""
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():
"""Get all user's as ``asyncpg.Record`` instances (default).
"""
users = await User.get()
for u in users:
print(u)
async def get_users_as_instances():
"""Get all user's and convert them all into ``User`` instances.
This can also be set as the default behavior if we would set
``_return_records`` to ``True`` on our ``User`` class.
"""
users = await User.get(records=False)
for u in users:
print(u)
async def get_foo_user():
"""Get user by the name of 'foo'. The ``get`` or ``get_one`` method's
accept **kwargs that will set a ``where`` clause on the query, to filter
the results.
``get`` always returns a list of objects, where ``get_one`` always returns
the first object.
"""
print(await User.get_one(name='foo'))
async def delete_all_users():
"""Delete the user's from the database.
"""
for u in await User.get(records=False):
print(f'Deleting user: {u}')
await u.delete()
async def main():
await orm.create_table(User)
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...')
await orm.drop_table(User)
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...