pgmodel

`app.contrib.sqlalchemy.pgmodel`

Un Módulo de Python que expone una clase base, para generar una instancia con metodos sencillos y gestionar a una Base de Datos PostgreSQL, usando psycopg.

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

`Base`

Bases: DeclarativeBase

Clase base para todos los modelos de SQLAlchemy, que utilizan DeclarativeBase.

Inherits

DeclarativeBase: La clase base de SQLAlchemy para modelos declarativos.

`BaseModel`

Esta clase proporciona una implementación por defecto para el atributo __tablename__, derivado del nombre de la clase. Esto Garantiza que los nombres de tabla se crean automáticamente basándose en los nombres de clase del modelo.

Attributes:

Name	Type	Description
`__tablename__`	`str`	El nombre de la tabla en la base de datos. Se establece automáticamente con el nombre de la clase en minúsculas.

`tablename()`

Extrae el nombre de la tabla, definido en el campo __tablename__, de la clase que se extiende de PgModel.

Returns:

Type	Description
`str`	El nombre de la tabla, derivado del nombre
`str`	de la clase, en minúsculas.

`PgModel`

Bases: BaseModel, Base

Genera una instancia con metodos sencillos para el manejo de PostgreSQL usando psycopg.

Inherits

BaseModel: Proporciona la generación automática de nombres de tabla basada en el nombre de la clase. Base: La clase base de SQLAlchemy para modelos declarativos, permitiendo definiciones de modelos.

Attributes:

Name	Type	Description
`_db`	`Optional[Session]`	donde almacena el objeto de la sesion de SQLAlchemy.
`_prefix`	`str`	El prefijo para las tablas.

Raises:

Type	Description
`TypeCheckError`	error de comprobación de tipos
`maximum recursion depth exceeded`	en caso de que no se pueda definir el valor de retorno de la propiedad `__tablename__`

Examples:

# se define una clase que extiende de PgModel
class Locales(PgModel):
    __tablename__: str = "Locales"

    id: Mapped[int] = mapped_column(primary_key=True)  # noqa: A003
    name: Mapped[str] = mapped_column(String(25))
    code: Mapped[Optional[str]] = mapped_column(String(12))
    status_id: Mapped[int] = mapped_column()
    created_at: Mapped[datetime.datetime] = mapped_column()
    updated_at: Mapped[datetime.datetime] = mapped_column()


# se crea una instancia de la clase
new_locale = Locales(request.app.pg_db)

# se asignan los valores
new_locale.name = body.name
new_locale.code = body.code
new_locale.status = DefaultStatus.Active

# se guardan los datos en la Base de Datos
# utilizando el método `save`
result = await new_locale.save()

`init(conn)`

Constructor de la clase PgModel.

Parameters:

Name	Type	Description	Default
`conn`	`Request`	una instancia de `Request.state` con el objeto de conexión a la Base de Datos	required

Raises:

Type	Description
`RuntimeError`	en caso de que no se pueda generar una conexión a la Base de Datos
`Exception`	cualquier otra Excepción al momento de crear la conexión

`get_all(select_fields=None, limit=0)`

Obtiene todos los registros de una tabla, adicionalmente permite una lista de campos a devolver y un limite para la cantidad de registros consultados.

Parameters:

Name	Type	Description	Default
`select_fields`	`list[str]`	una lista con los posibles campos de los registros a retornar. por defecto `None`.	`None`
`limit`	`int`	define el limite de registros a consultar. por defecto `0` (todos los registros).	`0`

Warning

Este metódo esta en revisión, puede cambiar en el futuro.

Notes

este método valida si cada campo, de la lista que se envia, existe en el modelo (objeto de tipo PgModel).

Returns:

Type	Description
`Optional[list[dict[str, Any]]]`	usualmente una lista de diccionarios con el resultado de la consulta, ó `None` en caso de error

Examples:

# ejemplo 1
...

resp = self.get_all()

# ejemplo 2
...

my_fields: list[str] = ["field1", "field2", "field3"]

await self.get_all(my_fields)

# ejemplo 3
...

await self.get_all(my_fields, 100)

# ejemplo 4
...

await self.get_all(limit=100)

`get_by_id(item_id, select_fields=None)`

Obtiene un registro de la Base de Datos por su ID, opcionalmente permite definir una lista de campos a devolver.

Parameters:

Name	Type	Description	Default
`item_id`	`int`	el id único del registro a consultar	required
`select_fields`	`list[str]`	una lista de campos a devolver, por defecto `None`.	`None`

Warning

Este metódo esta en revisión, puede cambiar en el futuro.

Notes

este método valida si cada campo, de la lista de campos que se envia, existe en el modelo (objeto de tipo PgModel).

Raises:

Type	Description
`SQLAlchemyError`	excepciones generadas por SQLAlchemy.

Returns:

Type	Description
`Optional[dict[str, Any]]`	un diccionario con el registro consultado, ó `None` en caso de error

Examples:

...
users = Users(request.app.pg_db)

# ejemplo 1
result = await users.get_by_id(user_id)

# ejemplo 2
result = await users.get_by_id(user_id, ["name", "avatar"])

`save()`

Guarda un objeto (de tipo PgModel) en la Base de Datos.

Raises:

Type	Description
`SQLAlchemyError`	excepciones generadas por SQLAlchemy.

Returns:

Type	Description
`Optional[int]`	el id del registro almacenado, ó `None` en caso de error

Examples:

...

new_user = Users(request.app.pg_db)

new_user.username = body.name
new_user.avatar = None

resp = await new_user.save()

`update()`

Actualiza un objeto (de tipo PgModel) en la Base de Datos.

Raises:

Type	Description
`SQLAlchemyError`	excepciones generadas por SQLAlchemy.

Returns:

Type	Description
`Optional[int]`	el id del registro actualizado, ó `None` en caso de error

Examples:

...

upd_user = Users(request.app.pg_db)

upd_user.username = body.name
upd_user.avatar = "/path/to/picture.png"

resp = await upd_user.update()

`remove(item_id)`

Elimina un registro de la Base de Datos.

Parameters:

Name	Type	Description	Default
`item_id`	`int`	el id único del registro a eliminar/actualizar (según el modo)	required

Raises:

Type	Description
`SQLAlchemyError`	excepciones generadas por SQLAlchemy.

Returns:

Type	Description
`Any`	`None` si el registro fue eliminado/actualizado con exito.

Examples:

1
2
3

...

result = await self.remove(body.id)

`to_dict()`

Convierte una instancia de PgModel en un diccionario.

Returns:

Type	Description
`Optional[dict[str, Any]]`	un diccionario con los campos y valores de la instancia ó `None` en caso de error

`custom_encoder(obj)`

Método utilizado por orjson para validar algunos tipos de datos (sobretodo fechas), mientras se decodifica los resultados de la consulta a la Base de Datos.

Parameters:

Name	Type	Description	Default
`obj`	`Any`	un objeto para serializar una subclase ó tipos arbitrarios	required

Notes

Esta es una función recursiva.
Esta función utiliza strftime para dar formato a las fechas con zona horaria.

Para más información, consulta:

Returns:

Type	Description
`Any`	un tipo soportado por `orjson` ó `None` en caso de error

Examples:

1
2
3

...
json_data = orjson.dumps(data, default=self.custom_encoder)
json_load = orjson.loads(json_data)

`raw_json(row=None)`

Convierte un objeto de tipo Row en un diccionario.

Notes

Esta método es de uso interno, utilizado por raw_query

Returns:

Type	Description
`Any`	un diccionario con los campos y valores del objeto tipo Row ó `None` en caso de error

pgmodel

app.contrib.sqlalchemy.pgmodel

Base

BaseModel

__tablename__()

PgModel

__init__(conn)

get_all(select_fields=None, limit=0)

get_by_id(item_id, select_fields=None)

save()

update()

remove(item_id)

to_dict()

custom_encoder(obj)

raw_json(row=None)

`app.contrib.sqlalchemy.pgmodel`

`Base`

`BaseModel`

`tablename()`

`PgModel`

`init(conn)`

`get_all(select_fields=None, limit=0)`

`get_by_id(item_id, select_fields=None)`

`save()`

`update()`

`remove(item_id)`

`to_dict()`

`custom_encoder(obj)`

`raw_json(row=None)`