fastack.controller¶

`Controller` ¶

Base Controller for creating REST APIs.

Examples:

from pydantic import BaseModel

class UserBody(BaseModel):
    name: str

class UserController(Controller):
    def get(self, id: int):
        user = User.get(id)
        return self.json("User", user)

    def post(self, body: UserBody):
        user = User.create(**body.dict())
        return self.json("Created", user)

    def put(self, id: int, body: UserBody):
        user = User.get(id)
        user.update(**body.dict())
        return self.json("Updated", user)

    def delete(self, id: int):
        user = User.get(id)
        user.delete()
        return self.json("Deleted")

user_controller = UserController()
app.include_controller(user_controller)

Attributes:

Name	Type	Description
`name`	`Optional[str]`	Name of the controller. If not provided, the name of the class will be used.
`url_prefix`	`Optional[str]`	URL prefix of the controller. If not provided, the name of the controller will be used.
`mapping_endpoints`	`Dict[str, str]`	Mapping to get default path.
`method_endpoints`	`Dict[str, str]`	Mapping to get default HTTP method.
`middlewares`	`Optional[Sequence[fastapi.params.Depends]]`	List of middlewares (dependencies) to be applied to all routes.
`json_response_class`	`Type[starlette.responses.JSONResponse]`	Class to be used for JSON responses.

Source code in fastack/controller.py

class Controller:
    """
    Base Controller for creating REST APIs.

    Example:

    ```python
    from pydantic import BaseModel

    class UserBody(BaseModel):
        name: str

    class UserController(Controller):
        def get(self, id: int):
            user = User.get(id)
            return self.json("User", user)

        def post(self, body: UserBody):
            user = User.create(**body.dict())
            return self.json("Created", user)

        def put(self, id: int, body: UserBody):
            user = User.get(id)
            user.update(**body.dict())
            return self.json("Updated", user)

        def delete(self, id: int):
            user = User.get(id)
            user.delete()
            return self.json("Deleted")

    user_controller = UserController()
    app.include_controller(user_controller)
    ```

    Attributes:
        name: Name of the controller. If not provided, the name of the class will be used.
        url_prefix: URL prefix of the controller. If not provided, the name of the controller will be used.
        mapping_endpoints: Mapping to get default path.
        method_endpoints: Mapping to get default HTTP method.
        middlewares: List of middlewares (dependencies) to be applied to all routes.
        json_response_class: Class to be used for JSON responses.

    """

    name: Optional[str] = None
    url_prefix: Optional[str] = None
    mapping_endpoints: Dict[str, str] = MAPPING_ENDPOINTS
    method_endpoints: Dict[str, str] = METHOD_ENDPOINTS
    middlewares: Optional[Sequence[params.Depends]] = []
    json_response_class: Type[JSONResponse] = JSONResponse

    def get_endpoint_name(self) -> str:
        """
        Get the name of the controller.
        This will be used to prefix the endpoint name (e.g user)
        when you create the absolute path of an endpoint using ``request.url_for``
        it looks like this ``request.url_for('user:get')``

        ``get`` here is the method name (responder) so you cDict[str, str]an replace it according to the method name
        such as ``post``, ``put``, ``delete``, ``retrieve``, etc.

        Returns:
            str: Name of the controller.
        """
        if self.name:
            return self.name

        c_suffix = "controller"
        name = type(self).__name__
        # remove "controller" text if any
        if len(name) > len(c_suffix):
            sfx = name[len(name) - len(c_suffix) :].lower()
            if sfx == c_suffix:
                name = name[: len(name) - len(c_suffix)]

        rv = ""
        for c in name:
            if c.isupper() and len(rv) > 0:
                c = "-" + c
            rv += c

        # Save endpoint name
        self.name = rv.lower()
        return name

    def url_for(self, name: str, **params: Dict[str, Any]) -> str:
        """
        Generate absolute URL for an endpoint.

        Args:
            name: Method name (e.g. retrieve).
            params: Can be path parameters or query parameters.
        """

        endpoint_name = self.join_endpoint_name(name)
        return url_for(endpoint_name, **params)

    def get_url_prefix(self) -> str:
        """
        Get the URL prefix of the controller.
        If not provided, the name of the controller will be used.

        Returns:
            str: URL prefix of the controller.
        """

        prefix = self.url_prefix
        if not prefix:
            prefix = self.get_endpoint_name()
        prefix = "/" + prefix.lstrip("/")
        return prefix

    def get_path(self, method: str) -> str:
        """
        Get the path of an endpoint.

        Args:
            method: Name of the method.
        """

        return self.mapping_endpoints.get(method) or ""

    def get_http_method(self, method: str) -> Union[str, None]:
        """
        Get the HTTP method of an endpoint.

        Args:
            method: Name of the method.
        """
        return self.method_endpoints.get(method) or None

    def join_endpoint_name(self, name: str) -> str:
        """
        Join endpoint name with controller name.

        Args:
            name: Name of the method.
        """

        return self.get_endpoint_name() + ":" + name

    def build(
        self,
        *,
        prefix: str = "",
        tags: Optional[List[str]] = None,
        dependencies: Optional[Sequence[params.Depends]] = None,
        default_response_class: Type[Response] = Default(JSONResponse),
        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
        callbacks: Optional[List[BaseRoute]] = None,
        routes: Optional[List[BaseRoute]] = None,
        redirect_slashes: bool = True,
        default: Optional[ASGIApp] = None,
        dependency_overrides_provider: Optional[Any] = None,
        route_class: Type[APIRoute] = APIRoute,
        on_startup: Optional[Sequence[Callable[[], Any]]] = None,
        on_shutdown: Optional[Sequence[Callable[[], Any]]] = None,
        deprecated: Optional[bool] = None,
        include_in_schema: bool = True,
    ) -> APIRouter:
        """
        Makes all APIs in controllers into a router (APIRouter)
        """

        endpoint_name = self.get_endpoint_name()
        tag_name = endpoint_name.replace("-", " ").title()
        if not tags:
            tags = [tag_name]

        if not prefix:
            prefix = self.get_url_prefix()

        if not dependencies:
            dependencies = []

        dependencies = dependencies + (self.middlewares or [])  # type: ignore[operator]
        router = APIRouter(
            prefix=prefix,
            tags=tags,
            dependencies=dependencies,
            default_response_class=default_response_class,
            responses=responses,
            callbacks=callbacks,
            routes=routes,
            redirect_slashes=redirect_slashes,
            default=default,
            dependency_overrides_provider=dependency_overrides_provider,
            route_class=route_class,
            on_startup=on_startup,
            on_shutdown=on_shutdown,
            deprecated=deprecated,
            include_in_schema=include_in_schema,
        )
        for method_name in dir(self):
            func = getattr(self, method_name)
            # skip if it's not a method, valid methods shouldn't be prefixed with _
            if method_name.startswith("_") or not isinstance(func, MethodType):
                continue

            http_method: Optional[str] = method_name.upper()
            if http_method not in HTTP_METHODS:
                http_method = self.get_http_method(method_name)

            # Checks if a method has an HTTP method.
            # Also, if no HTTP method is found there is another option to add the method to the router.
            # Just need to mark method using ``fastack.decorators.route()`` decorator with ``action=True`` parameter.
            is_action = getattr(func, "__route_action__", False)
            if http_method or is_action:
                # To generate an absolute path, using request.url_for(...)
                summary = f"{endpoint_name} {method_name.replace('_', ' ').title()}"
                default_path = self.get_path(method_name)
                params = getattr(func, "__route_params__", None) or {}
                if not params.get("methods", None):
                    params["methods"] = [http_method]

                name = params.pop("name", None)
                if not name:
                    name = method_name

                params["name"] = self.join_endpoint_name(name)

                if not params.get("summary", None):
                    params["summary"] = summary

                # if no path is provided, use the default path
                path = params.pop("path", None) or default_path
                router.add_api_route(path, func, **params)

        return router

    def serialize_data(self, obj: Any) -> Any:
        """
        Serialize data to JSON.
        By default it will use the "serialize" method on the object to convert the data.
        """

        func = getattr(obj, "serialize", None)
        if callable(func):
            obj = func()
        return obj

    def json(
        self,
        detail: str,
        data: Optional[Union[dict, list, object]] = None,
        *,
        status: int = 200,
        headers: Optional[dict] = None,
        allow_empty: bool = True,
        **kwargs: Any,
    ) -> JSONResponse:
        """
        Return a JSON response.
        By default the json response will be formatted like this:

        ```json
        {
            "detail": "...",
            "data": ...
        }
        ```

        Args:
            detail: Detail of the response.
            data: Data to be serialized.
            status: HTTP status code.
            headers: HTTP headers.
            allow_empty: Allows blank data to be shown to frontend.
            **kwargs (optional): Additional arguments to be passed to the JSONResponse.
        """

        content = {"detail": detail}
        if data or allow_empty:
            data = self.serialize_data(data)
            content["data"] = jsonable_encoder(data)

        return self.json_response_class(
            content, status_code=status, headers=headers, **kwargs
        )

`build(self, *, prefix='', tags=None, dependencies=None, default_response_class=<fastapi.datastructures.DefaultPlaceholder object at 0x7f0df36590d0>, responses=None, callbacks=None, routes=None, redirect_slashes=True, default=None, dependency_overrides_provider=None, route_class=<class 'fastapi.routing.APIRoute'>, on_startup=None, on_shutdown=None, deprecated=None, include_in_schema=True)` ¶

Makes all APIs in controllers into a router (APIRouter)

Source code in fastack/controller.py

def build(
    self,
    *,
    prefix: str = "",
    tags: Optional[List[str]] = None,
    dependencies: Optional[Sequence[params.Depends]] = None,
    default_response_class: Type[Response] = Default(JSONResponse),
    responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
    callbacks: Optional[List[BaseRoute]] = None,
    routes: Optional[List[BaseRoute]] = None,
    redirect_slashes: bool = True,
    default: Optional[ASGIApp] = None,
    dependency_overrides_provider: Optional[Any] = None,
    route_class: Type[APIRoute] = APIRoute,
    on_startup: Optional[Sequence[Callable[[], Any]]] = None,
    on_shutdown: Optional[Sequence[Callable[[], Any]]] = None,
    deprecated: Optional[bool] = None,
    include_in_schema: bool = True,
) -> APIRouter:
    """
    Makes all APIs in controllers into a router (APIRouter)
    """

    endpoint_name = self.get_endpoint_name()
    tag_name = endpoint_name.replace("-", " ").title()
    if not tags:
        tags = [tag_name]

    if not prefix:
        prefix = self.get_url_prefix()

    if not dependencies:
        dependencies = []

    dependencies = dependencies + (self.middlewares or [])  # type: ignore[operator]
    router = APIRouter(
        prefix=prefix,
        tags=tags,
        dependencies=dependencies,
        default_response_class=default_response_class,
        responses=responses,
        callbacks=callbacks,
        routes=routes,
        redirect_slashes=redirect_slashes,
        default=default,
        dependency_overrides_provider=dependency_overrides_provider,
        route_class=route_class,
        on_startup=on_startup,
        on_shutdown=on_shutdown,
        deprecated=deprecated,
        include_in_schema=include_in_schema,
    )
    for method_name in dir(self):
        func = getattr(self, method_name)
        # skip if it's not a method, valid methods shouldn't be prefixed with _
        if method_name.startswith("_") or not isinstance(func, MethodType):
            continue

        http_method: Optional[str] = method_name.upper()
        if http_method not in HTTP_METHODS:
            http_method = self.get_http_method(method_name)

        # Checks if a method has an HTTP method.
        # Also, if no HTTP method is found there is another option to add the method to the router.
        # Just need to mark method using ``fastack.decorators.route()`` decorator with ``action=True`` parameter.
        is_action = getattr(func, "__route_action__", False)
        if http_method or is_action:
            # To generate an absolute path, using request.url_for(...)
            summary = f"{endpoint_name} {method_name.replace('_', ' ').title()}"
            default_path = self.get_path(method_name)
            params = getattr(func, "__route_params__", None) or {}
            if not params.get("methods", None):
                params["methods"] = [http_method]

            name = params.pop("name", None)
            if not name:
                name = method_name

            params["name"] = self.join_endpoint_name(name)

            if not params.get("summary", None):
                params["summary"] = summary

            # if no path is provided, use the default path
            path = params.pop("path", None) or default_path
            router.add_api_route(path, func, **params)

    return router

`get_endpoint_name(self)` ¶

Get the name of the controller. This will be used to prefix the endpoint name (e.g user) when you create the absolute path of an endpoint using request.url_for it looks like this request.url_for('user:get')

get here is the method name (responder) so you cDict[str, str]an replace it according to the method name such as post, put, delete, retrieve, etc.

Returns:

Type	Description
`str`	Name of the controller.

Source code in fastack/controller.py

def get_endpoint_name(self) -> str:
    """
    Get the name of the controller.
    This will be used to prefix the endpoint name (e.g user)
    when you create the absolute path of an endpoint using ``request.url_for``
    it looks like this ``request.url_for('user:get')``

    ``get`` here is the method name (responder) so you cDict[str, str]an replace it according to the method name
    such as ``post``, ``put``, ``delete``, ``retrieve``, etc.

    Returns:
        str: Name of the controller.
    """
    if self.name:
        return self.name

    c_suffix = "controller"
    name = type(self).__name__
    # remove "controller" text if any
    if len(name) > len(c_suffix):
        sfx = name[len(name) - len(c_suffix) :].lower()
        if sfx == c_suffix:
            name = name[: len(name) - len(c_suffix)]

    rv = ""
    for c in name:
        if c.isupper() and len(rv) > 0:
            c = "-" + c
        rv += c

    # Save endpoint name
    self.name = rv.lower()
    return name

`get_http_method(self, method)` ¶

Get the HTTP method of an endpoint.

Parameters:

Name	Type	Description	Default
`method`	`str`	Name of the method.	required

Source code in fastack/controller.py

def get_http_method(self, method: str) -> Union[str, None]:
    """
    Get the HTTP method of an endpoint.

    Args:
        method: Name of the method.
    """
    return self.method_endpoints.get(method) or None

`get_path(self, method)` ¶

Get the path of an endpoint.

Parameters:

Name	Type	Description	Default
`method`	`str`	Name of the method.	required

Source code in fastack/controller.py

def get_path(self, method: str) -> str:
    """
    Get the path of an endpoint.

    Args:
        method: Name of the method.
    """

    return self.mapping_endpoints.get(method) or ""

`get_url_prefix(self)` ¶

Get the URL prefix of the controller. If not provided, the name of the controller will be used.

Returns:

Type	Description
`str`	URL prefix of the controller.

Source code in fastack/controller.py

def get_url_prefix(self) -> str:
    """
    Get the URL prefix of the controller.
    If not provided, the name of the controller will be used.

    Returns:
        str: URL prefix of the controller.
    """

    prefix = self.url_prefix
    if not prefix:
        prefix = self.get_endpoint_name()
    prefix = "/" + prefix.lstrip("/")
    return prefix

`join_endpoint_name(self, name)` ¶

Join endpoint name with controller name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the method.	required

Source code in fastack/controller.py

def join_endpoint_name(self, name: str) -> str:
    """
    Join endpoint name with controller name.

    Args:
        name: Name of the method.
    """

    return self.get_endpoint_name() + ":" + name

`json(self, detail, data=None, *, status=200, headers=None, allow_empty=True, **kwargs)` ¶

Return a JSON response. By default the json response will be formatted like this:

{
    "detail": "...",
    "data": ...
}

Parameters:

Name	Type	Description	Default
`detail`	`str`	Detail of the response.	required
`data`	`Union[dict, list, object]`	Data to be serialized.	`None`
`status`	`int`	HTTP status code.	`200`
`headers`	`Optional[dict]`	HTTP headers.	`None`
`allow_empty`	`bool`	Allows blank data to be shown to frontend.	`True`
`**kwargs`	`optional`	Additional arguments to be passed to the JSONResponse.	`{}`

Source code in fastack/controller.py

def json(
    self,
    detail: str,
    data: Optional[Union[dict, list, object]] = None,
    *,
    status: int = 200,
    headers: Optional[dict] = None,
    allow_empty: bool = True,
    **kwargs: Any,
) -> JSONResponse:
    """
    Return a JSON response.
    By default the json response will be formatted like this:

    ```json
    {
        "detail": "...",
        "data": ...
    }
    ```

    Args:
        detail: Detail of the response.
        data: Data to be serialized.
        status: HTTP status code.
        headers: HTTP headers.
        allow_empty: Allows blank data to be shown to frontend.
        **kwargs (optional): Additional arguments to be passed to the JSONResponse.
    """

    content = {"detail": detail}
    if data or allow_empty:
        data = self.serialize_data(data)
        content["data"] = jsonable_encoder(data)

    return self.json_response_class(
        content, status_code=status, headers=headers, **kwargs
    )

`serialize_data(self, obj)` ¶

Serialize data to JSON. By default it will use the "serialize" method on the object to convert the data.

Source code in fastack/controller.py

def serialize_data(self, obj: Any) -> Any:
    """
    Serialize data to JSON.
    By default it will use the "serialize" method on the object to convert the data.
    """

    func = getattr(obj, "serialize", None)
    if callable(func):
        obj = func()
    return obj

`url_for(self, name, **params)` ¶

Generate absolute URL for an endpoint.

Parameters:

Name	Type	Description	Default
`name`	`str`	Method name (e.g. retrieve).	required
`params`	`Dict[str, Any]`	Can be path parameters or query parameters.	`{}`

Source code in fastack/controller.py

def url_for(self, name: str, **params: Dict[str, Any]) -> str:
    """
    Generate absolute URL for an endpoint.

    Args:
        name: Method name (e.g. retrieve).
        params: Can be path parameters or query parameters.
    """

    endpoint_name = self.join_endpoint_name(name)
    return url_for(endpoint_name, **params)

`CreateController (Controller)` ¶

Controller for creating data.

Source code in fastack/controller.py

class CreateController(Controller):
    """
    Controller for creating data.
    """

    def create(self, body: BaseModel) -> Response:
        """
        adding a new object.
        """

        raise NotImplementedError  # pragma: no cover

`create(self, body)` ¶

adding a new object.

Source code in fastack/controller.py

def create(self, body: BaseModel) -> Response:
    """
    adding a new object.
    """

    raise NotImplementedError  # pragma: no cover

`CreateUpdateController (CreateController, UpdateController)` ¶

Controller for creating and updating data.

Source code in fastack/controller.py

class CreateUpdateController(CreateController, UpdateController):
    """
    Controller for creating and updating data.
    """

`DestroyController (Controller)` ¶

Controller for deleting data.

Source code in fastack/controller.py

class DestroyController(Controller):
    """
    Controller for deleting data.
    """

    def destroy(self, id: int) -> Response:
        """
        Delete an object with the given ID.
        """

        raise NotImplementedError  # pragma: no cover

`destroy(self, id)` ¶

Delete an object with the given ID.

Source code in fastack/controller.py

def destroy(self, id: int) -> Response:
    """
    Delete an object with the given ID.
    """

    raise NotImplementedError  # pragma: no cover

`ListController (Controller, ListControllerMixin)` ¶

Controller for listing data.

Attributes:

Name	Type	Description
`pagination_class`	`Type[fastack.pagination.Pagination]`	Class to be used for pagination.

Source code in fastack/controller.py

class ListController(Controller, ListControllerMixin):
    """
    Controller for listing data.

    Attributes:
        pagination_class: Class to be used for pagination.
    """

    def list(
        self, page: int = Query(1, gt=0), page_size: int = Query(10, gt=0)
    ) -> Response:
        """
        List data.
        """

        raise NotImplementedError  # pragma: no cover

`list(self, page=Query(1), page_size=Query(10))` ¶

List data.

Source code in fastack/controller.py

def list(
    self, page: int = Query(1, gt=0), page_size: int = Query(10, gt=0)
) -> Response:
    """
    List data.
    """

    raise NotImplementedError  # pragma: no cover

`ModelController (RetrieveController, CreateController, ListController, UpdateController, DestroyController)` ¶

Model Controller for creating REST APIs.

Source code in fastack/controller.py

class ModelController(
    RetrieveController,
    CreateController,
    ListController,
    UpdateController,
    DestroyController,
):
    """
    Model Controller for creating REST APIs.
    """

`ReadOnlyController (RetrieveController, ListController)` ¶

Controller for read-only data.

Source code in fastack/controller.py

class ReadOnlyController(RetrieveController, ListController):
    """
    Controller for read-only data.
    """

`RetrieveController (Controller)` ¶

Controller for retrieving data.

Source code in fastack/controller.py

class RetrieveController(Controller):
    """
    Controller for retrieving data.
    """

    def retrieve(self, id: int) -> Response:
        """
        Retrieve a single object with the given ID.
        """

        raise NotImplementedError  # pragma: no cover

`retrieve(self, id)` ¶

Retrieve a single object with the given ID.

Source code in fastack/controller.py

def retrieve(self, id: int) -> Response:
    """
    Retrieve a single object with the given ID.
    """

    raise NotImplementedError  # pragma: no cover

`UpdateController (Controller)` ¶

Controller for updating data.

Source code in fastack/controller.py

class UpdateController(Controller):
    """
    Controller for updating data.
    """

    def update(self, id: int, body: BaseModel) -> Response:
        """
        Update an object with the given ID.
        """

        raise NotImplementedError  # pragma: no cover

`update(self, id, body)` ¶

Update an object with the given ID.

Source code in fastack/controller.py

def update(self, id: int, body: BaseModel) -> Response:
    """
    Update an object with the given ID.
    """

    raise NotImplementedError  # pragma: no cover

Last update: January 17, 2022 13:26:42

fastack.controller¶

Controller ¶

get_endpoint_name(self) ¶

get_http_method(self, method) ¶

get_path(self, method) ¶

get_url_prefix(self) ¶

join_endpoint_name(self, name) ¶

json(self, detail, data=None, *, status=200, headers=None, allow_empty=True, **kwargs) ¶

serialize_data(self, obj) ¶

url_for(self, name, **params) ¶

CreateController (Controller) ¶

create(self, body) ¶

CreateUpdateController (CreateController, UpdateController) ¶

DestroyController (Controller) ¶

destroy(self, id) ¶

ListController (Controller, ListControllerMixin) ¶

list(self, page=Query(1), page_size=Query(10)) ¶

ModelController (RetrieveController, CreateController, ListController, UpdateController, DestroyController) ¶

ReadOnlyController (RetrieveController, ListController) ¶

RetrieveController (Controller) ¶

retrieve(self, id) ¶

UpdateController (Controller) ¶

update(self, id, body) ¶

`Controller` ¶

`get_endpoint_name(self)` ¶

`get_http_method(self, method)` ¶

`get_path(self, method)` ¶

`get_url_prefix(self)` ¶

`join_endpoint_name(self, name)` ¶

`json(self, detail, data=None, *, status=200, headers=None, allow_empty=True, **kwargs)` ¶

`serialize_data(self, obj)` ¶

`url_for(self, name, **params)` ¶

`CreateController (Controller)` ¶

`create(self, body)` ¶

`CreateUpdateController (CreateController, UpdateController)` ¶

`DestroyController (Controller)` ¶

`destroy(self, id)` ¶

`ListController (Controller, ListControllerMixin)` ¶

`list(self, page=Query(1), page_size=Query(10))` ¶

`ModelController (RetrieveController, CreateController, ListController, UpdateController, DestroyController)` ¶

`ReadOnlyController (RetrieveController, ListController)` ¶

`RetrieveController (Controller)` ¶

`retrieve(self, id)` ¶

`UpdateController (Controller)` ¶

`update(self, id, body)` ¶