dynamic_conn

`app.contrib.sqlalchemy.dynamic_conn`

Módulo de Python para gestionar conexiones dinámicas a PostgreSQL(psycopg) y MySQL(pymysql)

Para más información, te recomendamos leer la documentación del Proyecto.

`dyn_connect(db_list, conn_name, conn_uri, ping=True)`

Establece la conexión a una Base datos de forma dinámica.

Warning

Esta función no es asíncrona.

Es responsabilidad de desarollador:

Gestionar los nombres de las conexiones, para evitar crear la mismas conexiones, con diferentes nombres.
Cerrar la sesión(sessionmaker) al finalizar las consultas.

Parameters:

Name	Type	Description	Default
`db_list`	`list[Any]`	el objeto `databases` de `request.app` para almacenar las conexiones activas	required
`conn_name`	`str`	el nombre de la conexion a almacenar	required
`conn_uri`	`str`	la url de conexion, para la base de datos	required
`ping`	`bool`	permite hacer un ping a la Base de Datos, para comprobar el estado de la conexión, por defecto `True`	`True`

Raises:

Type	Description
`DefaultException`	excepciones para el manejo de errores personalizados
`Exception`	fallo al conectarse a la Base de Datos
`None`	en caso de un error

Returns:

Type	Description
`Optional[Any]`	un objeto con la sesión(sessionmaker), ó `None` en caso de error

Examples:

from app.contrib.sqlalchemy.dynamic_conn import dyn_connect
from sqlalchemy.orm.session import Session
...

# lifespan
fastapp.databases = {}

...

uri: str = "postgresql+psycopg2://awesome_conn_uri"

db: Optional[Session] = dyn_connect(request.app.databases, "awesome_name_conn", uri, True)

result: Any = db.execute(text("SELECT * from users"))
resp: Optional[Any] = None = result.fetchall()

...

try:
    db.close()
except Exception as e:
    logger.warning("error while closing the session")