Reference

`cattle_grid.extensions`

`BaseConfig`

Bases: BaseModel

Source code in cattle_grid/extensions/__init__.py

class BaseConfig(BaseModel): ...

`Extension` `dataclass`

Data model for an extension

Parameters:

Name	Type	Description	Default
`name`	`str`		required
`module`	`str`		required
`lifespan`	`Callable[list, Awaitable[None]] \| None`		`None`
`config_class`	`Any`		`<class 'cattle_grid.extensions.BaseConfig'>`
`Config`	`Any \| None`		`None`
`configuration`	`Any \| None`		`None`
`activity_router`	`RabbitRouter`	Includable to RabbitBroker router.	`<dynamic>`
`api_router`	`APIRouter`	`APIRouter` class, used to group path operations, for example to structure an app in multiple files. It would then be included in the `FastAPI` app, or in another `APIRouter` (ultimately included in the app). Read more about it in the FastAPI docs for Bigger Applications - Multiple Files. Example `from fastapi import APIRouter, FastAPI app = FastAPI() router = APIRouter() @router.get("/users/", tags=["users"]) async def read_users(): return [{"username": "Rick"}, {"username": "Morty"}] app.include_router(router)`	`<dynamic>`
`api_prefix`	`str`		`''`
`transformer`	`Callable[list, Awaitable[Dict]] \| None`		`None`
`transformer_inputs`	`List[str] \| None`		`None`
`transformer_outputs`	`List[str] \| None`		`None`
`lookup_method`	`Callable[list, Awaitable[Dict]] \| None`		`None`
`lookup_order`	`int \| None`		`None`
`method_information`	`List[MethodInformation]`	Built-in mutable sequence. If no argument is given, the constructor creates a new empty list. The argument must be an iterable if specified.	`<dynamic>`

Source code in cattle_grid/extensions/__init__.py

@dataclass
class Extension:
    """Data model for an extension"""

    name: str
    """name of the extension, must be unique"""

    module: str
    """module the extension is defined in, should be set to `__name__`"""

    lifespan: Callable[[], Awaitable[None]] | None = None
    """The lifespan function"""

    config_class: Any = BaseConfig
    """Expected configuration class"""

    Config: Any | None = None
    """Annotation to retrieve the configuration, e.g.

    ```python
    @extension.lookup()
    async def lookup(
        lookup: Lookup, config: extension.Config
    ) -> Lookup: ...
    ```
    """
    configuration: Any | None = None

    activity_router: RabbitRouter = field(default_factory=RabbitRouter)
    api_router: APIRouter = field(default_factory=APIRouter)
    api_prefix: str = ""

    transformer: Callable[[Dict], Awaitable[Dict]] | None = None
    transformer_inputs: List[str] | None = None
    transformer_outputs: List[str] | None = None

    lookup_method: Callable[[str], Awaitable[Dict]] | None = None
    lookup_order: int | None = None

    method_information: List[MethodInformation] = field(default_factory=list)

    def __post_init__(self):
        def get_config():
            return self.configuration

        self.Config = Annotated[self.config_class, Depends(get_config)]

    def configure(self, config: dict):
        """Configures the extension

        The configuration is validated using the config_class.
        """
        self.configuration = self.config_class.model_validate(config)

    def transform(self, inputs: List[str] = [], outputs: List[str] = []):
        """Allows building the extension via decorator. Usage:

        ```python
        extension = Extension("my extension")

        @extension.transform(inputs=["inputs"], outputs=["outputs"])
        async def transformer(a: dict):
            ...
        ```
        """
        if self.transformer:
            raise ValueError("You should not override an existing transformer")

        def inner(func):
            self.transformer = func
            self.transformer_inputs = inputs
            self.transformer_outputs = outputs

            return func

        return inner

    def subscribe(self, routing_key: str, description: str | None = None):
        """Allows building the extension via decorator.

        ```python
        extension = Extension("my extension")

        @extension.subscribe("routing_key")
        async def subscriber(msg: dict): ...
        ```

        Dependency injection is available for the subscriber function.
        """

        def inner(func):
            if description is None:
                function_description = func.__doc__
            else:
                function_description = description

            self.activity_router.subscriber(
                routing_key, exchange=exchange(), title=routing_key
            )(func)

            if not skip_method_information(routing_key):

                self.method_information.append(
                    MethodInformation(
                        module=self.module,
                        routing_key=routing_key,
                        description=function_description,
                    )
                )

            return func

        return inner

    def lookup(self):
        """Allows building the extension via decorator.

        ```python
        extension = Extension("my extension")

        @extension.lookup()
        async def lookup(l: Lookup) -> Lookup:
            ...
        ```
        Dependency injection is available for the lookup function.
        """

        def inner(func):
            self.lookup_method = func
            return func

        return inner

    def get(self, path):
        """Allows one to add a get endpoint to the API Router
        of the extension

        """

        def inner(func):
            self.api_router.get(path)(func)
            return func

        return inner

`Config = None` `class-attribute` `instance-attribute`

Annotation to retrieve the configuration, e.g.

@extension.lookup()
async def lookup(
    lookup: Lookup, config: extension.Config
) -> Lookup: ...

`config_class = BaseConfig` `class-attribute` `instance-attribute`

Expected configuration class

`lifespan = None` `class-attribute` `instance-attribute`

The lifespan function

`module` `instance-attribute`

module the extension is defined in, should be set to __name__

`name` `instance-attribute`

name of the extension, must be unique

`configure(config)`

Configures the extension

The configuration is validated using the config_class.

Source code in cattle_grid/extensions/__init__.py

def configure(self, config: dict):
    """Configures the extension

    The configuration is validated using the config_class.
    """
    self.configuration = self.config_class.model_validate(config)

`get(path)`

Allows one to add a get endpoint to the API Router of the extension

Source code in cattle_grid/extensions/__init__.py

def get(self, path):
    """Allows one to add a get endpoint to the API Router
    of the extension

    """

    def inner(func):
        self.api_router.get(path)(func)
        return func

    return inner

`lookup()`

Allows building the extension via decorator.

extension = Extension("my extension")

@extension.lookup()
async def lookup(l: Lookup) -> Lookup:
    ...

Dependency injection is available for the lookup function.

Source code in cattle_grid/extensions/__init__.py

def lookup(self):
    """Allows building the extension via decorator.

    ```python
    extension = Extension("my extension")

    @extension.lookup()
    async def lookup(l: Lookup) -> Lookup:
        ...
    ```
    Dependency injection is available for the lookup function.
    """

    def inner(func):
        self.lookup_method = func
        return func

    return inner

`subscribe(routing_key, description=None)`

Allows building the extension via decorator.

extension = Extension("my extension")

@extension.subscribe("routing_key")
async def subscriber(msg: dict): ...

Dependency injection is available for the subscriber function.

Source code in cattle_grid/extensions/__init__.py

def subscribe(self, routing_key: str, description: str | None = None):
    """Allows building the extension via decorator.

    ```python
    extension = Extension("my extension")

    @extension.subscribe("routing_key")
    async def subscriber(msg: dict): ...
    ```

    Dependency injection is available for the subscriber function.
    """

    def inner(func):
        if description is None:
            function_description = func.__doc__
        else:
            function_description = description

        self.activity_router.subscriber(
            routing_key, exchange=exchange(), title=routing_key
        )(func)

        if not skip_method_information(routing_key):

            self.method_information.append(
                MethodInformation(
                    module=self.module,
                    routing_key=routing_key,
                    description=function_description,
                )
            )

        return func

    return inner

`transform(inputs=[], outputs=[])`

Allows building the extension via decorator. Usage:

extension = Extension("my extension")

@extension.transform(inputs=["inputs"], outputs=["outputs"])
async def transformer(a: dict):
    ...

Source code in cattle_grid/extensions/__init__.py

def transform(self, inputs: List[str] = [], outputs: List[str] = []):
    """Allows building the extension via decorator. Usage:

    ```python
    extension = Extension("my extension")

    @extension.transform(inputs=["inputs"], outputs=["outputs"])
    async def transformer(a: dict):
        ...
    ```
    """
    if self.transformer:
        raise ValueError("You should not override an existing transformer")

    def inner(func):
        self.transformer = func
        self.transformer_inputs = inputs
        self.transformer_outputs = outputs

        return func

    return inner