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)

Estás leyendo una vista previa.

Inicia sesión con Google para leer la página completa. Una cuenta de Google incluye 5 páginas gratuitas en total; el alumnado y el profesorado leen las páginas de su curso sin límite.

Iniciar sesión