applipy-pg 0.2.0

An applipy library for working with PostgreSQL.

Applipy PostgreSQL

An applipy library for working with PostgreSQL.

It lets you declare connections in the configuration of your application that get turned into postgres connection pools that can be accessed by declaring the dependency in your classes.

The connection pools are created the first time they are used and closed on application shutdown.

Usage

You can define connections to databases in you application config file:

# dev.yaml
app:
  name: demo
  modules:
  - applipy_pg.PgModule

pg:
  connections:
  # Defines an anonimous db connection pool
  - user: username
    host: mydb.local
    port: 5432
    dbname: demo
    password: $3cr37
  # Defines an db connection pool with name "db2"
  - name: db2
    user: username
    host: mydb.local
    port: 5432
    dbname: demo
    password: $3cr37

The configuration definition above defines two database connection pools. These can be accessed through applipy's dependency injection system:

from applipy_pg import PgPool

class DoSomethingOnDb:
    def __init__(self, pool: PgPool) -> None:
        self._pool = pool

    async def do_something(self) -> None:
        async with self.pool.cursor() as cur:
            # cur is a aiopg.Cursor
            await cur.execute('SELECT 1')
            await cur.fetchone()

from typing import Annotated
from applipy_inject import name

class DoSomethingOnDb2:
    def __init__(self, pool: Annotated[PgPool, name('db2')]) -> None:
        self._pool = pool

    async def do_something(self) -> None:
        async with self.pool.cursor() as cur:
            # cur is a aiopg.Cursor
            await cur.execute('SELECT 2')
            await cur.fetchone()

The aiopg.Pool instance can be accessed using the PgPool.pool() method.

Each connection pool can be further configured by setting a config attribute with a dict containing the extra paramenters to be passed to aiopg.create_pool():

pg:
  connections:
  - user: username
    host: mydb.local
    port: 5432
    dbname: demo
    password: $3cr37
    config:
      minsize: 5
      timeout: 100.0

You can also define a global configuration that will serve as a base to all database connections defined by setting pg.global_config.

pg:
  global_config:
    minsize: 5
    timeout: 100.0
  connections:
  # ...

Migrations

This library also includes a migrations functionality. How to use it:

First, define your migrations:

class DemoMigrationSubject_20240101(ClassNameMigration):
    def __init__(self, pool: PgPool) -> None:
        self._pool = pool  # Import whatever resources you need

    async def migrate(self) -> None:
        # Do you migrations...
        async with self._pool.cursor() as cur:
            ...

If you want more control over how the version and subject of the migration is defined, you can extend Migration and implement your own logic.

Then, create your migrations module:

class MyMigrationsModule(Module):
    def configure(self, bind: BindFunction, register: RegisterFunction) -> None:
        bind(Migration, DemoMigrationSubject_20240101)

    @classmethod
    def depends_on(cls) -> tuple[type[Module], ...]:
        return PgMigrationsModule,

Finally, you can optionally set the name of the connection to use for the migrations audit table. This table is used to know what migrations have been run and which migrations should be ran:

pg:
  connections:
  # Defines an db connection pool with name "db2"
  - name: db2
    # ...
  migrations:
    # sets the connection named "db2" as the connection to use for the
    # migrations audit table
    connection: db2