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

Encara no has enllaçat el compte? Entra un cop a l'Escola amb el Moodle del centre.