Skip to content

Manager

sensei.Manager

Manager(sync_client=None, async_client=None, *, required=True)

This class serves as a bridge between the application and Sensei, to dynamically provide a client for routed function calls. It separately stores httpx.AsyncClient and httpx.Client. To use Manager, you need to create it and pass it to the router.

Import it directly from Sensei:

from sensei import Manager
Example
from sensei import Manager, Router, Client

manager = Manager()
router = Router('httpx://example-api.com', manager=manager)

@router.get('/users/{id_}')
def get_user(id_: int) -> User:
    pass

with Client(base_url=router.base_url) as client:
    manager.set(client)
    user = get_user(1)
    print(user)
    manager.pop()
PARAMETER DESCRIPTION
sync_client

An instance of httpx.Client.

TYPE: Client DEFAULT: None

async_client

An instance of httpx.AsyncClient.

TYPE: AsyncClient DEFAULT: None

required

Whether to throw the error in get if a client is not set

TYPE: bool DEFAULT: True

RAISES DESCRIPTION
TypeError

If the provided client is not an instance of AsyncClient or Client.

Source code in sensei/client/manager.py
11
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
def __init__(
        self,
        sync_client: Optional[Client] = None,
        async_client: Optional[AsyncClient] = None,
        *,
        required: bool = True,
) -> None:
    """
    This class serves as a bridge between the application and Sensei, to dynamically provide a client for
    routed function calls.
    It separately stores `httpx.AsyncClient` and `httpx.Client`.
    To use `Manager`, you need to create it and pass it to the router.

    Import it directly from Sensei:

    ```python
    from sensei import Manager
    ```

    Example:
        ```python
        from sensei import Manager, Router, Client

        manager = Manager()
        router = Router('httpx://example-api.com', manager=manager)

        @router.get('/users/{id_}')
        def get_user(id_: int) -> User:
            pass

        with Client(base_url=router.base_url) as client:
            manager.set(client)
            user = get_user(1)
            print(user)
            manager.pop()
        ```

    Args:
        sync_client (Client): An instance of `httpx.Client`.
        async_client (AsyncClient): An instance of `httpx.AsyncClient`.
        required (bool): Whether to throw the error in `get` if a client is not set

    Raises:
        TypeError: If the provided client is not an instance of AsyncClient or Client.
    """
    self._sync_client = self._validate_client(sync_client, True)
    self._async_client = self._validate_client(async_client, True, True)
    self._required = required

set

set(client)

Set a client instance in the manager if no client is currently set. If a client of the provided type is already set, it raises a CollectionLimitError.

Example
from sensei import Manager, Router, Client, AsyncClient

manager = Manager()
router = Router('httpx://example-api.com', manager=manager)

client = Client(base_url=router.base_url)
aclient = AsyncClient(base_url=router.base_url)

manager.set(client)
manager.set(aclient)
PARAMETER DESCRIPTION
client

The client instance to set.

TYPE: BaseClient

RAISES DESCRIPTION
CollectionLimitError

If a client of the provided type is already set.

TypeError

If the provided client is not an instance of AsyncClient or Client.

Source code in sensei/client/manager.py
 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
def set(self, client: BaseClient) -> None:
    """
    Set a client instance in the manager if no client is currently set. If a client of the provided type is
    already set, it raises a `CollectionLimitError`.

    Example:
        ```python
        from sensei import Manager, Router, Client, AsyncClient

        manager = Manager()
        router = Router('httpx://example-api.com', manager=manager)

        client = Client(base_url=router.base_url)
        aclient = AsyncClient(base_url=router.base_url)

        manager.set(client)
        manager.set(aclient)
        ```

    Args:
        client (BaseClient): The client instance to set.

    Raises:
        CollectionLimitError: If a client of the provided type is already set.
        TypeError: If the provided client is not an instance of AsyncClient or Client.
    """
    is_async = isinstance(client, AsyncClient)
    set_client = self._get_client(is_async)

    if set_client is None:
        if is_async:
            self._async_client = self._validate_client(client, is_async=True)
        else:
            self._sync_client = self._validate_client(client)
    else:
        raise CollectionLimitError(self.__class__, [(1, Client), (1, AsyncClient)])

pop

pop(is_async=False)

Remove and return the currently set client.

Example
manager = Manager()

manager = Manager(sync_client=client, async_client=aclient)
client = manager.pop(is_async=False)
aclient = manager.pop(is_async=True)
print(client, aclient)
PARAMETER DESCRIPTION
is_async

Whether client instance is async

TYPE: bool DEFAULT: False

RETURNS DESCRIPTION
BaseClient

The client instance that was managed.

TYPE: BaseClient

RAISES DESCRIPTION
AttributeError

If no client is set.

Source code in sensei/client/manager.py
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 pop(self, is_async: bool = False) -> BaseClient:
    """
    Remove and return the currently set client.

    Example:
        ```python
        manager = Manager()

        manager = Manager(sync_client=client, async_client=aclient)
        client = manager.pop(is_async=False)
        aclient = manager.pop(is_async=True)
        print(client, aclient)
        ```

    Args:
        is_async (bool): Whether client instance is async

    Returns:
        BaseClient: The client instance that was managed.

    Raises:
        AttributeError: If no client is set.
    """
    set_client = self._get_client(is_async, True, True)
    return set_client

empty

empty(is_async=False)

Check if the manager has a client of the provided type set.

Example
manager = Manager()

manager = Manager(sync_client=client)
manager.pop()
print(manager.empty()) # Output: True
PARAMETER DESCRIPTION
is_async

Whether client instance is async

TYPE: bool DEFAULT: False

RETURNS DESCRIPTION
bool

True if no client is set, False otherwise.

TYPE: bool

Source code in sensei/client/manager.py
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def empty(self, is_async: bool = False) -> bool:
    """
    Check if the manager has a client of the provided type set.

    Example:
        ```python
        manager = Manager()

        manager = Manager(sync_client=client)
        manager.pop()
        print(manager.empty()) # Output: True
        ```

    Args:
        is_async (bool): Whether client instance is async

    Returns:
        bool: True if no client is set, False otherwise.
    """
    set_client = self._get_client(is_async)

    return set_client is None

get

get(is_async=False)

Retrieve the currently set client of the provided type. This method returns the managed client without removing it.

Example
manager = Manager()

manager = Manager(sync_client=client, async_client=aclient)
client = manager.get(is_async=False)
aclient = manager.get(is_async=True)
print(client, aclient)
PARAMETER DESCRIPTION
is_async

Whether client instance is async

TYPE: bool DEFAULT: False

RETURNS DESCRIPTION
BaseClient

The client instance that is being managed.

TYPE: BaseClient

RAISES DESCRIPTION
AttributeError

If no client is set.

Source code in sensei/client/manager.py
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
def get(self, is_async: bool = False) -> BaseClient:
    """
    Retrieve the currently set client of the provided type.
    This method returns the managed client without removing it.

    Example:
        ```python
        manager = Manager()

        manager = Manager(sync_client=client, async_client=aclient)
        client = manager.get(is_async=False)
        aclient = manager.get(is_async=True)
        print(client, aclient)
        ```

    Args:
        is_async (bool): Whether client instance is async

    Returns:
        BaseClient: The client instance that is being managed.

    Raises:
        AttributeError: If no client is set.
    """
    set_client = self._get_client(is_async, required=self._required)

    return set_client