Skip to content

Reference

cattle_grid.extensions

BaseConfig

Bases: BaseModel

Source code in cattle_grid/extensions/__init__.py 
21
class BaseConfig(BaseModel): ...

Extension dataclass

Data model for an extension

Parameters:

Name Type Description Default
name str
required
module str
required
description str | None

description of the extension. If not set is populated from the docstring of the extension

 None
lifespan Callable[list, AbstractAsyncContextManager[None]] | None
None
config_class Any
<class 'cattle_grid.extensions.BaseConfig'>
Config Any | None
None
ConfigFastAPI Any | None
None
configuration Any | None
None
activity_router RabbitRouter

Includable to RabbitBroker router.

 <dynamic>
api_router APIRouter

API 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[Lookup]] | None
None
lookup_order int | None
None
method_information List[MethodInformationModel]

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 
 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
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
@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__`"""

    description: str | None = field(
        default=None,
        metadata={
            "description": "description of the extension. If not set is populated from the docstring of the extension"
        },
    )

    lifespan: Callable[[Any], AbstractAsyncContextManager[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: ...
    ```
    """
    ConfigFastAPI: Any | None = None
    """Annotation to retrieve the configuration using FastAPI Depends"""

    configuration: Any | None = None

    activity_router: RabbitRouter = field(default_factory=RabbitRouter)
    api_router: APIRouter = field(
        default_factory=APIRouter, metadata={"description": "API router"}
    )
    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[[Lookup], Awaitable[Lookup]] | None = None
    lookup_order: int | None = None

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

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

        self.Config = Annotated[self.config_class, Depends(get_config)]
        self.ConfigFastAPI = Annotated[self.config_class, FastAPIDepends(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(message: 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=global_container.exchange, title=routing_key
            )(func)

            if not skip_method_information(routing_key):
                self.method_information.append(
                    MethodInformationModel(
                        module=self.module,
                        routing_key=routing_key,
                        description=function_description,
                    )
                )

            return func

        return inner

    def subscribe_on_account_exchange(self, routing_key: str):
        """Allows building the extension via decorator.

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

        def inner(func):
            self.activity_router.subscriber(
                routing_key,
                exchange=global_container.account_exchange,
                title=routing_key,
            )(func)

            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, **kwargs):
        """Allows one to add a get endpoint to the API Router
        of the extension

        Usage:

        ```python
        @extension.get("/path")
        async def get_endpoint():
            pass

        ```
        """

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

        return inner

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

        Usage:

        ```python
        @extension.post("/path")
        async def post_endpoint():
            pass

        ```
        """

        def inner(func):
            self.api_router.post(path, **kwargs)(func)
            return func

        return inner

    def include_router(self, router: APIRouter, **kwargs):
        """Includes the router as an api router"""

        self.api_router.include_router(router, **kwargs)

Config class-attribute instance-attribute 

Config: Any | None = None

Annotation to retrieve the configuration, e.g.

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

ConfigFastAPI class-attribute instance-attribute 

ConfigFastAPI: Any | None = None

Annotation to retrieve the configuration using FastAPI Depends

config_class class-attribute instance-attribute 

config_class: Any = BaseConfig

Expected configuration class

lifespan class-attribute instance-attribute 

lifespan: (
    Callable[[Any], AbstractAsyncContextManager[None]]
    | None
) = None

The lifespan function

module instance-attribute 

module: str

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

name instance-attribute 

name: str

name of the extension, must be unique

configure 

configure(config: dict)

Configures the extension

The configuration is validated using the config_class.

Source code in cattle_grid/extensions/__init__.py 
84
85
86
87
88
89
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 

get(path, **kwargs)

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

Usage:

@extension.get("/path")
async def get_endpoint():
    pass
Source code in cattle_grid/extensions/__init__.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
def get(self, path, **kwargs):
    """Allows one to add a get endpoint to the API Router
    of the extension

    Usage:

    ```python
    @extension.get("/path")
    async def get_endpoint():
        pass

    ```
    """

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

    return inner

include_router 

include_router(router: APIRouter, **kwargs)

Includes the router as an api router

Source code in cattle_grid/extensions/__init__.py 
226
227
228
229
def include_router(self, router: APIRouter, **kwargs):
    """Includes the router as an api router"""

    self.api_router.include_router(router, **kwargs)

lookup 

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 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
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

post 

post(path, **kwargs)

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

Usage:

@extension.post("/path")
async def post_endpoint():
    pass
Source code in cattle_grid/extensions/__init__.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
def post(self, path, **kwargs):
    """Allows one to add a post endpoint to the API Router
    of the extension

    Usage:

    ```python
    @extension.post("/path")
    async def post_endpoint():
        pass

    ```
    """

    def inner(func):
        self.api_router.post(path, **kwargs)(func)
        return func

    return inner

subscribe 

subscribe(routing_key: str, description: str | None = None)

Allows building the extension via decorator.

extension = Extension("my extension")

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

Dependency injection is available for the subscriber function.

Source code in cattle_grid/extensions/__init__.py 
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
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(message: 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=global_container.exchange, title=routing_key
        )(func)

        if not skip_method_information(routing_key):
            self.method_information.append(
                MethodInformationModel(
                    module=self.module,
                    routing_key=routing_key,
                    description=function_description,
                )
            )

        return func

    return inner

subscribe_on_account_exchange 

subscribe_on_account_exchange(routing_key: str)

Allows building the extension via decorator.

Dependency injection is available for the subscriber function.

Source code in cattle_grid/extensions/__init__.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def subscribe_on_account_exchange(self, routing_key: str):
    """Allows building the extension via decorator.

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

    def inner(func):
        self.activity_router.subscriber(
            routing_key,
            exchange=global_container.account_exchange,
            title=routing_key,
        )(func)

        return func

    return inner

transform 

transform(inputs: List[str] = [], outputs: List[str] = [])

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 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
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

cattle_grid.extensions.load

add_routers_to_broker 

add_routers_to_broker(
    broker: RabbitBroker, extensions: List[Extension]
)

Adds the routers to the broker

Source code in cattle_grid/extensions/load/__init__.py 
102
103
104
105
106
107
def add_routers_to_broker(broker: RabbitBroker, extensions: List[Extension]):
    """Adds the routers to the broker"""

    for extension in extensions:
        if extension.activity_router:
            broker.include_router(extension.activity_router)

add_routes_to_api 

add_routes_to_api(
    app: FastAPI, extensions: List[Extension]
)

Adds the routes to the api

Source code in cattle_grid/extensions/load/__init__.py 
110
111
112
113
114
115
def add_routes_to_api(app: FastAPI, extensions: List[Extension]):
    """Adds the routes to the api"""
    for extension in extensions:
        if extension.api_router:
            if extension.api_prefix:
                app.include_router(extension.api_router, prefix=extension.api_prefix)

build_lookup 

build_lookup(extensions: List[Extension]) -> LookupMethod

Builds the lookup method

Source code in cattle_grid/extensions/load/__init__.py 
71
72
73
74
75
76
77
78
79
80
81
82
def build_lookup(extensions: List[Extension]) -> LookupMethod:
    """Builds the lookup method"""
    methods = ordered_lookups(extensions)

    async def lookup_result(lookup: Lookup) -> Lookup:
        for method in methods:
            lookup = await inject(method)(lookup)
            if lookup.result is not None:
                return lookup
        return lookup

    return lookup_result

build_transformer 

build_transformer(
    extensions: List[Extension],
) -> Callable[[Dict], Awaitable[Dict]]

Build the transformer

Source code in cattle_grid/extensions/load/__init__.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
def build_transformer(extensions: List[Extension]) -> Callable[[Dict], Awaitable[Dict]]:
    """Build the transformer"""
    transformers = get_transformers(extensions)
    steps = transformation_steps(transformers)

    async def transformer(data: dict):
        for step in steps:
            for plugin in step:
                data.update(await plugin.transformer(data))

        return data

    return transformer

collect_method_information 

collect_method_information(
    extensions: List[Extension],
) -> List[MethodInformationModel]

Collects the method information from the extensions

Source code in cattle_grid/extensions/load/__init__.py 
118
119
120
121
122
def collect_method_information(
    extensions: List[Extension],
) -> List[MethodInformationModel]:
    """Collects the method information from the extensions"""
    return sum((extension.method_information for extension in extensions), [])

lifespan_from_extensions async 

lifespan_from_extensions(extensions: List[Extension])

Creates the lifespan from the extensions

Source code in cattle_grid/extensions/load/__init__.py 
93
94
95
96
97
98
99
@asynccontextmanager
async def lifespan_from_extensions(extensions: List[Extension]):
    """Creates the lifespan from the extensions"""
    lifespans = collect_lifespans(extensions)

    async with iterate_lifespans(lifespans):
        yield

load_extension 

load_extension(extension_information: dict) -> Extension

Loads a single extension

Source code in cattle_grid/extensions/load/__init__.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def load_extension(extension_information: dict) -> Extension:
    """Loads a single extension"""
    module_name = extension_information.get("module")

    if module_name is None:
        raise ValueError("module is required")

    module = importlib.import_module(module_name)
    extension = module.extension
    extension.configure(extension_information.get("config", {}))

    if extension.description is None and module.__doc__ is not None:
        extension.description = module.__doc__

    if "lookup_order" in extension_information:
        extension.lookup_order = extension_information["lookup_order"]
    if "api_prefix" in extension_information:
        extension.api_prefix = extension_information["api_prefix"]

    return extension

load_extensions 

load_extensions(settings) -> List[Extension]

Loads the extensions from settings

Source code in cattle_grid/extensions/load/__init__.py 
43
44
45
46
47
48
49
50
51
52
53
def load_extensions(settings) -> List[Extension]:
    """Loads the extensions from settings"""

    extensions = [
        load_extension(extension_information)
        for extension_information in settings.extensions
    ]

    logger.info("Loaded extensions: %s", ", ".join(f"'{e.name}'" for e in extensions))

    return extensions

set_globals 

set_globals(extensions: List[Extension])

Sets global variables in cattle_grid.dependencies

Source code in cattle_grid/extensions/load/__init__.py 
85
86
87
88
89
90
def set_globals(extensions: List[Extension]):
    """Sets global variables in cattle_grid.dependencies"""
    from cattle_grid.dependencies.globals import global_container

    global_container.transformer = build_transformer(extensions)
    global_container.lookup = build_lookup(extensions)