Escribe para buscar…

PostgreSQL

Introducción

Para acceder a una base de dados puedes utilizar un ORM o ejecutar sentencias SQL directamente.

El ecosistema asyncio tiene varias bibliotecas cliente de Postgres, de las cuales asyncpg y aiopg son las más populares.

asyncpg tiene una API agradable y es la de mayor rendimiento, pero no implementa PEP-249, sino que prefiere el estilo paramatrizado de Postgres para evitar que asyncpg interprete o reescriba la consulta

Específicamente, esto significa que los parámetros tienen el prefijo $, por ejemplo:

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

Esto es problemático ya que es muy fácil confundir el orden de las variables, por ejemplo, este tipo de error,

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

mientras que un estilo de parámetro con nombre, por ejemplo

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

Es mucho más difícil equivocarse.

Para habilitar un estilo de parámetro con nombre, podemod utilizar Databases que envuelven asyncpg permitiendo cláusulas de texto de SQLAlchemy, como se usa en el fragmento anterior.

Configuración básica

Recomiendo este simple fragmento para configurar una conexión de bases de datos con 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)

lo que permite usos como,

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

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

con todos los métodos principales (fetch_one, fetch_all, execute, execute_many transacciones y opciones de conexión compatibles . Por ejemplo, (siguiendo el fragmento 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_},
        )

Avanzado; conversión de tipo

asyncpg admite la conversión de tipos personalizados entre los tipos de Postgres y Python. Por ejemplo, una columna JSON en la base de datos se puede volcar automáticamente a la base de datos y cargar desde ella, o una enumeración se puede convertir de una enumeración de Python a la base de datos y regresar cuando se carga. Por ejemplo, si tenemos esta estructura de base de datos,

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

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

y ejecutar consultas como,

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},
)

Sería genial si esto funcionara y eso result["details"] fuera un dict y result["state"] fuera una instancia de TrafficLight. Esto es posible definiendo cómo codificar y decodificar tipos hacia y desde tipos de Postgres usando un códec de tipos,

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",
    )

    ... # Run queries as above

Sin embargo, esto es una molestia, ya que los tipos de códecs deben configurarse cada vez que se utiliza una conexión. En su lugar init, se puede utilizar el argumento asyncpg para inicializar la conexión. Al poner esto junto con el ejemplo básico, se obtiene,

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 luego permite usos como,

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

Referencias