Escriu per cercar…

PostgreSQL

Introducció

Per accedir a una base de dades pots utilitzar un ORM o executar sentències SQL directament.

L’ecosistema asyncio té diverses biblioteques client de Postgres, de les quals asyncpg i aiopg són les més populars.

asyncpg té una API agradable i és la de més rendiment, però no implementa PEP-249, sinó que prefereix l’estil parametritzat de Postgres per evitar que asyncpg interpreti o reescrigui la consulta.

En concret, això vol dir que els paràmetres porten el prefix $, per exemple:

python
await conn.execute("SELECT * FROM table WHERE id = $1", id)

Això és problemàtic, perquè és molt fàcil confondre l’ordre de les variables. Per exemple, aquest tipus d’error,

python
await conn.execute(
    "SELECT * FROM table WHERE id = $1 AND active = $2",
     active,
     id,
)

mentre que amb un estil de paràmetre amb nom, per exemple

python
await conn.execute(
    "SELECT * FROM table WHERE id = :id AND active = :active",
    {"id": id, "active": active},
)

és molt més difícil equivocar-se.

Per habilitar l’estil de paràmetre amb nom, pots utilitzar Databases, que embolcalla asyncpg i permet clàusules de text de SQLAlchemy, tal com es fa al fragment anterior.

Configuració bàsica

Et recomano aquest fragment senzill per configurar una connexió a la base de dades amb Quart,

python
from typing import Any, Optional

from databases import Database
from quart import Quart

class QuartDatabases:
    def __init__(self, app: Optional[Quart] = None, **db_args: Any) -> None:
        self._db_args = db_args
        if app is not None:
            self.init_app(app)

    def init_app(self, app: Quart) -> None:
        self._url = app.config["QUART_DATABASES_URI"]
        app.before_serving(self._before_serving)
        app.after_serving(self._after_serving)

    async def _before_serving(self) -> None:
        self._db = Database(url=self._url, **self._db_args)
        await self._db.connect()

    async def _after_serving(self) -> None:
        await self._db.disconnect()

    def __getattr__(self, name: str) -> Any:
        return getattr(self._db, name)

que permet usos com ara,

python
app = Quart(__name__)
db = QuartDatabases(app)

@app.route("/")
async def index():
    return await db.fetch_val("SELECT COUNT(*) FROM mytable")

amb tots els mètodes principals (fetch_one, fetch_all, execute, execute_many), transaccions i opcions de connexió compatibles. Per exemple (seguint el fragment anterior),

python
@app.route("/<int:id_>/", methods=["POST"])
async def index(id_: int):
    data = await request.get_json()
    async with db.connection() as connection
        await connection.fetch_val("SELECT COUNT(*) FROM mytable")
        await connection.execute(
            "UPDATE mytable SET clm = :val WHERE id = :id",
            values={"val": data["clm"], "id": id_},
        )

Avançat: conversió de tipus

asyncpg admet la conversió de tipus personalitzats entre els tipus de Postgres i els de Python. Per exemple, una columna JSON de la base de dades es pot bolcar i carregar automàticament, o una enumeració es pot convertir d’una enumeració de Python a la base de dades i tornar quan es carrega. Per exemple, si tens aquesta estructura de base de dades,

sql
CREATE TYPE TRAFFIC_LIGHT_T AS ENUM ('RED', 'AMBER', 'GREEN');

CREATE TABLE lights (
    id SERIAL PRIMARY KEY,
    details JSONB,
    state TRAFFIC_LIGHT_T
);

i executes consultes com ara,

python
from enum import Enum

class TrafficLight(Enum):
    RED = "RED"
    AMBER = "AMBER"
    GREEN = "GREEN"

result = await db.fetch_one("SELECT details, state FROM lights LIMIT 1")
await db.execute(
    "INSERT INTO lights (details, state) VALUES (:details, :state)",
    values={"details": {"location": "London"}, "state": TrafficLight.RED},
)

Estaria molt bé que això funcionés i que result["details"] fos un dict i result["state"] fos una instància de TrafficLight. És possible si defineixes com codificar i descodificar tipus cap als tipus de Postgres i des d’ells, mitjançant un còdec de tipus,

python
import json

async with db.connection() as connection:
    await connection.raw_connection.set_type_code(
        "jsonb",
         encoder=json.dumps,
         decoder=json.loads,
         schema="pg_catalog",
    )

    await connection.raw_connection.set_type_code(
        "traffic_light_t",
        encoder=lambda type_: type_.value,
        decoder=TrafficLight
        schema="public",
        format="text",
    )

    ... # Executa les consultes com abans

Això, però, és empipador: has de configurar els còdecs cada cop que utilitzes una connexió. En comptes d’això pots fer servir l’argument init d’asyncpg per inicialitzar la connexió. Si ho ajuntes amb l’exemple bàsic, obtens,

python
from typing import Any, Callable, Optional

from databases import Database
from quart import Quart


class QuartDatabases:
    def __init__(self, app: Optional[Quart] = None, **db_args: Any) -> None:
        self._db_args = db_args
        self._codecs = []
        if app is not None:
            self.init_app(app)

    def init_app(self, app: Quart) -> None:
        self._url = app.config["QUART_DATABASES_URI"]
        app.before_serving(self._before_serving)
        app.after_serving(self._after_serving)

    def set_type_codec(
        self,
        type_: str,
        encoder: Callable,
        decoder: Callable,
        schema: Optional[str] = None,
        format: Optional[str] = None,
    ) -> None:
        self._codecs.append(type_, encoder, decoder, schema, format)

    async def _init(self, connection: asyncpg.Connection) -> None:
        for type_, encoder, decoder, schema, format in self._codecs:
            await connection.set_type_code(
                type_, encoder, decoder, schema, format
            )

    async def _before_serving(self) -> None:
        self._db = Database(url=self._url, init=self._init, **self._db_args)
        await self._db.connect()

    async def _after_serving(self) -> None:
        await self._db.disconnect()

    def __getattr__(self, name: str) -> Any:
        return getattr(self._db, name)

que després permet usos com ara,

python
import json
from enum import Enum

class TrafficLight(Enum):
    RED = "RED"
    AMBER = "AMBER"
    GREEN = "GREEN"

app = Quart(__name__)
db = QuartDatabases(app)

db.set_type_codec(
    "jsonb",
     encoder=json.dumps,
     decoder=json.loads,
     schema="pg_catalog",
)

db.set_type_code(
    "traffic_light_t",
    encoder=lambda type_: type_.value,
    decoder=TrafficLight
    schema="public",
    format="text",
)

@app.route("/lights/<int:id_>/")
async def index(id_: int):
    return await db.fetch_val(
        "SELECT details, state FROM lights WHERE id = :id",
        values={"id": id_},
    )

Quart-DB

Referències