SQLAlchemyBaseController

Extiende BaseController con métodos CRUD listos para SQLAlchemy async.

fastapi_basekit.aio.sqlalchemy.controller.base.SQLAlchemyBaseController

Bases: BaseController

BaseController para SQLAlchemy (AsyncSession).

Controlador base específico para proyectos que usan SQLAlchemy con async/await. Incluye soporte para joins, ordenamiento personalizado y operadores OR en filtros, características específicas de SQL.

Source code in fastapi_basekit/aio/sqlalchemy/controller/base.py

class SQLAlchemyBaseController(BaseController):
    """BaseController para SQLAlchemy (AsyncSession).

    Controlador base específico para proyectos que usan SQLAlchemy con
    async/await. Incluye soporte para joins, ordenamiento personalizado
    y operadores OR en filtros, características específicas de SQL.
    """

    service: BaseService = Depends()

    # Campos adicionales a excluir específicos de SQLAlchemy
    _params_excluded_fields: ClassVar[Set[str]] = {
        "self",
        "page",
        "count",
        "search",
        "use_or",
        "joins",
        "order_by",
        "__class__",
        "args",
        "kwargs",
        "id",
        "payload",
        "data",
        "validated_data",
    }

    async def list(
        self,
        *,
        use_or: bool = False,
        joins: Optional[List[str]] = None,
        order_by: Optional[Any] = None,
    ):
        """
        Lista registros con paginación usando SQLAlchemy.

        Args:
            use_or: Si True, usa OR en lugar de AND para los filtros
            joins: Lista de relaciones a hacer JOIN eager loading
            order_by: Expresión de ordenamiento (ej: User.created_at.desc())
        """
        params = self._params(skip_frames=2)
        service_params = {
            **params,
            "use_or": use_or,
            "joins": joins,
        }
        items, total = await self.service.list(**service_params)
        count = params.get("count") or 0
        total_pages = (total + count - 1) // count if count > 0 else 0
        pagination = {
            "page": params.get("page"),
            "count": count,
            "total": total,
            "total_pages": total_pages,
        }
        return self.format_response(data=items, pagination=pagination)

    async def retrieve(self, id: str, *, joins: Optional[List[str]] = None):
        """
        Obtiene un registro por ID.

        Args:
            id: ID del registro
            joins: Lista de relaciones a hacer JOIN eager loading
        """
        item = await self.service.retrieve(id, joins=joins)
        return self.format_response(data=item)

    async def create(
        self,
        validated_data: Any,
        *,
        check_fields: Optional[List[str]] = None,
    ):
        """
        Crea un nuevo registro.

        Args:
            validated_data: Datos validados para crear
            check_fields: Campos a verificar por duplicados antes de crear
        """
        result = await self.service.create(validated_data, check_fields)
        return self.format_response(result, message="Creado exitosamente")

    def to_dict(self, obj: Any):
        """Convierte un modelo SQLAlchemy a dict."""
        if hasattr(obj, "model_dump"):
            return obj.model_dump()
        # Para modelos SQLAlchemy que usan __dict__
        if hasattr(obj, "__dict__"):
            return {
                k: v for k, v in obj.__dict__.items() if not k.startswith("_")
            }
        return obj

Functions

list(*, use_or=False, joins=None, order_by=None) async

Lista registros con paginación usando SQLAlchemy.

Parameters:

Name	Type	Description	Default
use_or	bool	Si True, usa OR en lugar de AND para los filtros	False
joins	Optional[List[str]]	Lista de relaciones a hacer JOIN eager loading	None
order_by	Optional[Any]	Expresión de ordenamiento (ej: User.created_at.desc())	None

Source code in fastapi_basekit/aio/sqlalchemy/controller/base.py

async def list(
    self,
    *,
    use_or: bool = False,
    joins: Optional[List[str]] = None,
    order_by: Optional[Any] = None,
):
    """
    Lista registros con paginación usando SQLAlchemy.

    Args:
        use_or: Si True, usa OR en lugar de AND para los filtros
        joins: Lista de relaciones a hacer JOIN eager loading
        order_by: Expresión de ordenamiento (ej: User.created_at.desc())
    """
    params = self._params(skip_frames=2)
    service_params = {
        **params,
        "use_or": use_or,
        "joins": joins,
    }
    items, total = await self.service.list(**service_params)
    count = params.get("count") or 0
    total_pages = (total + count - 1) // count if count > 0 else 0
    pagination = {
        "page": params.get("page"),
        "count": count,
        "total": total,
        "total_pages": total_pages,
    }
    return self.format_response(data=items, pagination=pagination)

retrieve(id, *, joins=None) async

Obtiene un registro por ID.

Parameters:

Name	Type	Description	Default
id	str	ID del registro	required
joins	Optional[List[str]]	Lista de relaciones a hacer JOIN eager loading	None

Source code in fastapi_basekit/aio/sqlalchemy/controller/base.py

async def retrieve(self, id: str, *, joins: Optional[List[str]] = None):
    """
    Obtiene un registro por ID.

    Args:
        id: ID del registro
        joins: Lista de relaciones a hacer JOIN eager loading
    """
    item = await self.service.retrieve(id, joins=joins)
    return self.format_response(data=item)

create(validated_data, *, check_fields=None) async

Crea un nuevo registro.

Parameters:

Name	Type	Description	Default
validated_data	Any	Datos validados para crear	required
check_fields	Optional[List[str]]	Campos a verificar por duplicados antes de crear	None

Source code in fastapi_basekit/aio/sqlalchemy/controller/base.py

async def create(
    self,
    validated_data: Any,
    *,
    check_fields: Optional[List[str]] = None,
):
    """
    Crea un nuevo registro.

    Args:
        validated_data: Datos validados para crear
        check_fields: Campos a verificar por duplicados antes de crear
    """
    result = await self.service.create(validated_data, check_fields)
    return self.format_response(result, message="Creado exitosamente")

to_dict(obj)

Convierte un modelo SQLAlchemy a dict.

Source code in fastapi_basekit/aio/sqlalchemy/controller/base.py

def to_dict(self, obj: Any):
    """Convierte un modelo SQLAlchemy a dict."""
    if hasattr(obj, "model_dump"):
        return obj.model_dump()
    # Para modelos SQLAlchemy que usan __dict__
    if hasattr(obj, "__dict__"):
        return {
            k: v for k, v in obj.__dict__.items() if not k.startswith("_")
        }
    return obj

Métodos heredables

async def list(self, *, use_or=False, joins=None, order_by=None) -> BasePaginationResponse:
    """Lista paginada con filtros desde query string."""

async def retrieve(self, id: str, *, joins=None) -> BaseResponse:
    """Retrieve por ID."""

async def create(self, validated_data, *, check_fields=None) -> BaseResponse:
    """Create con duplicate check."""

update y delete se heredan de BaseController.

Patrón canónico

from fastapi_basekit.aio.sqlalchemy.controller.base import SQLAlchemyBaseController
from fastapi_basekit.schema.base import BasePaginationResponse, BaseResponse

@cbv(router)
class ThingController(SQLAlchemyBaseController):
    service: ThingService = Depends(get_thing_service)
    schema_class = ThingResponseSchema

    @router.get("/", response_model=BasePaginationResponse[ThingResponseSchema])
    async def list_things(self, page: int = Query(1), count: int = Query(10)):
        return await self.list()

    @router.get("/{id}", response_model=BaseResponse[ThingResponseSchema])
    async def get_thing(self, id: uuid.UUID):
        return await self.retrieve(id)

    @router.post("/", response_model=BaseResponse[ThingResponseSchema], status_code=201)
    async def create_thing(self, data: ThingCreateSchema):
        return await self.create(data)

Patrón completo