Saltar a contenido

BaseController

Controller base agnóstico de ORM.

fastapi_basekit.aio.controller.base.BaseController

Montar rutas CRUD genericas y captura errores de negocio.

Source code in fastapi_basekit/aio/controller/base.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
class BaseController:
    """Montar rutas CRUD genericas y captura errores de negocio."""

    service = Depends()
    schema_class: ClassVar[Type[BaseModel]]
    action: ClassVar[Optional[str]] = None
    request: Request
    _params_excluded_fields: ClassVar[Set[str]] = {
        "self",
        "page",
        "count",
        "search",
        "order_by",
        "__class__",
        "args",
        "kwargs",
        "id",
        "payload",
        "data",
        "validated_data",
    }

    def __init__(self) -> None:
        endpoint_func = (
            self.request.scope.get("endpoint")
            if hasattr(self, "request") and self.request
            else None
        )
        self.action = endpoint_func.__name__ if endpoint_func else None

    def get_schema_class(self) -> Type[BaseModel]:
        assert self.schema_class is not None, (
            "'%s' should either include a `schema_class` attribute, "
            "or override the `get_serializer_class()` method."
            % self.__class__.__name__
        )
        return self.schema_class

    async def check_permissions_class(self):
        permissions = self.check_permissions()
        if permissions:
            for permission in permissions:
                obj = permission()
                check = await obj.has_permission(self.request)
                if not check:
                    raise PermissionException(obj.message_exception)

    def check_permissions(self) -> List[Type[BasePermission]]:
        pass

    async def list(self):
        params = self._params()
        items, total = await self.service.list(**params)
        count = params.get("count") or 0
        page = params.get("page") or 1

        total_pages = (total + count - 1) // count if count > 0 else 0
        pagination = {
            "page": page,
            "count": count,
            "total": total,
            "total_pages": total_pages,
        }
        return self.format_response(data=items, pagination=pagination)

    async def retrieve(self, id: str):
        item = await self.service.retrieve(id)
        return self.format_response(data=item)

    async def create(self, validated_data: Any):
        result = await self.service.create(validated_data)
        return self.format_response(result, message="Creado exitosamente")

    async def update(self, id: str, validated_data: Any):
        result = await self.service.update(id, validated_data)
        return self.format_response(result, message="Actualizado exitosamente")

    async def delete(self, id: str):
        await self.service.delete(id)
        return self.format_response(None, message="Eliminado exitosamente")

    def format_response(
        self,
        data: Any,
        pagination: Optional[Dict[str, Any]] = None,
        message: Optional[str] = None,
        response_status: str = "success",
    ) -> BaseModel:
        schema = self.get_schema_class()

        if isinstance(data, list):
            data_dicts = [self.to_dict(item) for item in data]
            adapter = TypeAdapter(List[schema])
            data_parsed = adapter.validate_python(data_dicts)
        elif self.service.repository and isinstance(
            data, self.service.repository.model
        ):
            data_parsed = self.to_dict(data)
            data_parsed = schema.model_validate(data_parsed)
        elif isinstance(data, dict):
            data_parsed = schema.model_validate(data)
        else:
            data_parsed = data

        if pagination:
            return BasePaginationResponse(
                data=data_parsed,
                pagination=pagination,
                message=message or "Operación exitosa",
                status=response_status,
            )
        else:
            return BaseResponse(
                data=data_parsed,
                message=message or "Operación exitosa",
                status=response_status,
            )

    def _params(self, skip_frames: int = 1) -> Dict[str, Any]:
        """
        Extrae parámetros automáticamente usando introspección.

        Usa query_params como fuente de verdad para determinar QUÉ parámetros
        existen, y luego intenta obtener sus VALORES validados desde el frame
        del método llamador (con tipos ya convertidos por FastAPI).

        Args:
            skip_frames: Número de frames a saltar (1 por defecto para
                llamadas directas, 2 para controllers heredados)
        """
        # Obtener query_params como fuente de verdad
        query_params = self.request.query_params if self.request else {}

        # Parámetros especiales de paginación y búsqueda
        standard_params = {"page", "count", "search", "order_by"}

        # Valores por defecto
        page = 1
        count = 10
        search = None
        order_by = None
        filters = {}

        # Intentar obtener valores validados del frame local
        frame = inspect.currentframe()
        caller_locals = {}

        if frame:
            # Navegar hacia atrás en la pila según skip_frames
            caller_frame = frame
            for _ in range(skip_frames):
                if caller_frame and caller_frame.f_back:
                    caller_frame = caller_frame.f_back
                else:
                    break

            if caller_frame:
                caller_locals = caller_frame.f_locals

        # Procesar cada parámetro de query_params
        for param_name, param_value in query_params.items():
            # Intentar obtener valor validado del frame local
            validated_value = caller_locals.get(param_name)

            # Si no existe en locals, usar el valor del query_param
            final_value = (
                validated_value if validated_value is not None else param_value
            )

            # Clasificar el parámetro
            if param_name == "page":
                page = (
                    int(final_value)
                    if not isinstance(final_value, int)
                    else final_value
                )
            elif param_name == "count":
                count = (
                    int(final_value)
                    if not isinstance(final_value, int)
                    else final_value
                )
            elif param_name == "search":
                search = final_value
            elif param_name == "order_by":
                order_by = final_value
            elif param_name not in standard_params:
                # Es un filtro
                filters[param_name] = final_value

        return {
            "page": page,
            "count": count,
            "search": search,
            "order_by": order_by,
            "filters": filters,
        }

    def to_dict(self, obj: Any):
        if hasattr(obj, "model_dump"):
            return obj.model_dump()
        return obj

Functions

format_response(data, pagination=None, message=None, response_status='success')

Source code in fastapi_basekit/aio/controller/base.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def format_response(
    self,
    data: Any,
    pagination: Optional[Dict[str, Any]] = None,
    message: Optional[str] = None,
    response_status: str = "success",
) -> BaseModel:
    schema = self.get_schema_class()

    if isinstance(data, list):
        data_dicts = [self.to_dict(item) for item in data]
        adapter = TypeAdapter(List[schema])
        data_parsed = adapter.validate_python(data_dicts)
    elif self.service.repository and isinstance(
        data, self.service.repository.model
    ):
        data_parsed = self.to_dict(data)
        data_parsed = schema.model_validate(data_parsed)
    elif isinstance(data, dict):
        data_parsed = schema.model_validate(data)
    else:
        data_parsed = data

    if pagination:
        return BasePaginationResponse(
            data=data_parsed,
            pagination=pagination,
            message=message or "Operación exitosa",
            status=response_status,
        )
    else:
        return BaseResponse(
            data=data_parsed,
            message=message or "Operación exitosa",
            status=response_status,
        )

get_schema_class()

Source code in fastapi_basekit/aio/controller/base.py 
42
43
44
45
46
47
48
def get_schema_class(self) -> Type[BaseModel]:
    assert self.schema_class is not None, (
        "'%s' should either include a `schema_class` attribute, "
        "or override the `get_serializer_class()` method."
        % self.__class__.__name__
    )
    return self.schema_class

check_permissions()

Source code in fastapi_basekit/aio/controller/base.py 
59
60
def check_permissions(self) -> List[Type[BasePermission]]:
    pass

check_permissions_class() async

Source code in fastapi_basekit/aio/controller/base.py 
50
51
52
53
54
55
56
57
async def check_permissions_class(self):
    permissions = self.check_permissions()
    if permissions:
        for permission in permissions:
            obj = permission()
            check = await obj.has_permission(self.request)
            if not check:
                raise PermissionException(obj.message_exception)

to_dict(obj)

Source code in fastapi_basekit/aio/controller/base.py 
210
211
212
213
def to_dict(self, obj: Any):
    if hasattr(obj, "model_dump"):
        return obj.model_dump()
    return obj

_params(skip_frames=1)

Extrae parámetros automáticamente usando introspección.

Usa query_params como fuente de verdad para determinar QUÉ parámetros existen, y luego intenta obtener sus VALORES validados desde el frame del método llamador (con tipos ya convertidos por FastAPI).

Parameters:

Name Type Description Default
skip_frames int

Número de frames a saltar (1 por defecto para llamadas directas, 2 para controllers heredados)

 1
Source code in fastapi_basekit/aio/controller/base.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def _params(self, skip_frames: int = 1) -> Dict[str, Any]:
    """
    Extrae parámetros automáticamente usando introspección.

    Usa query_params como fuente de verdad para determinar QUÉ parámetros
    existen, y luego intenta obtener sus VALORES validados desde el frame
    del método llamador (con tipos ya convertidos por FastAPI).

    Args:
        skip_frames: Número de frames a saltar (1 por defecto para
            llamadas directas, 2 para controllers heredados)
    """
    # Obtener query_params como fuente de verdad
    query_params = self.request.query_params if self.request else {}

    # Parámetros especiales de paginación y búsqueda
    standard_params = {"page", "count", "search", "order_by"}

    # Valores por defecto
    page = 1
    count = 10
    search = None
    order_by = None
    filters = {}

    # Intentar obtener valores validados del frame local
    frame = inspect.currentframe()
    caller_locals = {}

    if frame:
        # Navegar hacia atrás en la pila según skip_frames
        caller_frame = frame
        for _ in range(skip_frames):
            if caller_frame and caller_frame.f_back:
                caller_frame = caller_frame.f_back
            else:
                break

        if caller_frame:
            caller_locals = caller_frame.f_locals

    # Procesar cada parámetro de query_params
    for param_name, param_value in query_params.items():
        # Intentar obtener valor validado del frame local
        validated_value = caller_locals.get(param_name)

        # Si no existe en locals, usar el valor del query_param
        final_value = (
            validated_value if validated_value is not None else param_value
        )

        # Clasificar el parámetro
        if param_name == "page":
            page = (
                int(final_value)
                if not isinstance(final_value, int)
                else final_value
            )
        elif param_name == "count":
            count = (
                int(final_value)
                if not isinstance(final_value, int)
                else final_value
            )
        elif param_name == "search":
            search = final_value
        elif param_name == "order_by":
            order_by = final_value
        elif param_name not in standard_params:
            # Es un filtro
            filters[param_name] = final_value

    return {
        "page": page,
        "count": count,
        "search": search,
        "order_by": order_by,
        "filters": filters,
    }

Atributos de clase

Atributo Tipo Default Descripción
service Depends() required Service inyectado
schema_class Type[BaseModel] required Schema para serializar response
action Optional[str] None Auto-set a request.scope["endpoint"].__name__
request Request injected FastAPI request

Comportamiento

  • __init__ lee endpoint → puebla self.action
  • format_response() valida data contra get_schema_class() y wrappea en BaseResponse o BasePaginationResponse
  • _params(skip_frames=2) extrae page/count/search/order_by + filtros del request

Ejemplo

from fastapi_basekit.aio.controller.base import BaseController

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

    @router.get("/")
    async def list_things(self):
        items, total = await self.service.list()
        return self.format_response(
            items,
            pagination={"page": 1, "count": 10, "total": total, "total_pages": 1},
        )

Para SQLAlchemy con joins/CRUD inherited, usa SQLAlchemyBaseController.