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)

Estàs llegint una vista prèvia.

Inicia sessió amb Google per llegir la pàgina completa.

Inicia sessió amb Google

Amb qualsevol compte de Google. Només et demanarem que acceptis les condicions del servei.