Escriu per cercar…

PostgreSQL

Aquesta pàgina encara no s'ha traduït — es mostra en l'idioma original:Español

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 llegint una vista prèvia.

Inicia sessió amb Google per llegir la pàgina completa. Un compte de Google inclou 5 pàgines gratuïtes en total; l'alumnat i el professorat llegeixen les pàgines del seu curs sense límit.

Inicia sessió