Preparers/Finalizers
When you need to handle a response in a nonstandard way or add or change arguments before a request, you can apply preparers and finalizers respectively. Before the start, we need to remember about hook levels.
Hook Levels (Order)¶
As was mentioned in Hook Levels, there are three ways of applying converters:
Example
from sensei import Router, camel_case, snake_case
router = Router(
'https://api.example.com',
body_case=camel_case,
response_case=snake_case
)
@router.post('/users')
def create_user(first_name: str, birth_city: str, ...) -> User:
pass
from sensei import Router, camel_case, snake_case
router = Router('https://api.example.com')
@router.post('/users', body_case=camel_case, response_case=snake_case)
def create_user(first_name: str, birth_city: str, ...) -> User:
pass
router = Router(host, response_case=camel_case)
class User(APIModel):
def __header_case__(self, s: str) -> str:
return kebab_case(s)
@staticmethod
def __response_case__(s: str) -> str:
return snake_case(s)
@classmethod
@router.get('/users/{id_}')
def get(cls, id_: Annotated[int, Path(alias='id')]) -> Self: pass
These levels are not only related to case converters. They are also related to other hooks, such as preparers and finalizers.
But here they don't determine the single Preparer/Finalizer from all that will be executed (determine priority), like for case converters. Each of them will be executed but in a different order. So, this type of hook levels is called Order Levels.
Here is an illustration of the execution orders of Preparers/Finalizers.
flowchart TB
RoutedModel["Routed Model Level (First Order)"] --> |replaces| Router["Router Level (First Order)"]
Router["Router Level (First Order)"] --> Route["Route Level (Second Order)"]
Route --> result{{result}}
style RoutedModel fill:#C94A2D,stroke:#000,stroke-width:2px;
style Router fill:#D88C00,stroke:#000,stroke-width:2px;
style Route fill:#2F8B5D,stroke:#000,stroke-width:2px;
Preparers¶
Preparer is a function that takes an instance of Args
as the argument and returns the modified.
Preparers are used for request preparation. That means adding or changing arguments
before request.
Preparers are executed after internal argument parsing. So, all request parameters are available in
Args
model within a preparer.
Algorithm¶
Look at the example below:
from sensei import Args, Router, APIModel
from pydantic import NonNegativeInt, EmailStr
router = Router('https://api.example.com')
class User(APIModel):
id: NonNegativeInt
email: EmailStr
nickname: str
@router.get('/users/{id_}')
def get_user(id_: int) -> User:
pass
@get_user.prepare
def _get_user_in(args: Args) -> Args:
print(f'Preparing arguments for {args.url} request')
return args
user = get_user(1)
print(user)
from sensei import Args, Router, APIModel
from pydantic import NonNegativeInt, EmailStr
def _prepare_args(args: Args) -> Args:
print(f'Preparing arguments for {args.url} request')
return args
router = Router('https://api.example.com', __prepare_args__=_prepare_args)
class User(APIModel):
id: NonNegativeInt
email: EmailStr
nickname: str
@router.get('/users/{id_}')
def get_user(id_: int) -> User:
pass
user = get_user(1)
print(user)
from sensei import Args, Router, APIModel
from pydantic import NonNegativeInt, EmailStr
router = Router('https://api.example.com')
class User(APIModel):
id: NonNegativeInt
email: EmailStr
nickname: str
@staticmethod
def __prepare_args__(args: Args) -> Args:
print(f'Preparing arguments for {args.url} request')
return args
@classmethod
@router.get('/users/{id_}')
def get(cls, id_: int) -> "User":
pass
user = User.get(1)
print(user)
Preparing arguments for /users/1 request
User(id=1, email="[email protected]", nickname="john")
Let's take this code step by step.
Step 1: Define Route¶
First, we need to define a route(s)
@router.get('/users/{id_}')
def get_user(id_: int) -> User:
pass
@router.get('/users/{id_}')
def get_user(id_: int) -> User:
pass
@classmethod
@router.get('/users/{id_}')
def get(cls, id_: int) -> "User":
pass
Step 2: Define Preparer¶
To define a preparer, you have to create a function, that takes only one argument of Args
type and returns
a value of the same type. The next step is based on a level type:
Decorate it as @<routed_function>.prepare
@get_user.prepare
def _get_user_in(args: Args) -> Args:
print(f'Preparing arguments for {args.url} request')
return args
Pass the function to Router
object
def _prepare_args(args: Args) -> Args:
print(f'Preparing arguments for {args.url} request')
return args
router = Router('https://api.example.com', __prepare_args__=_prepare_args)
You only need to define a class or static method with the name __prepare_args__
inside User
. Sensei will use this
hook when it's necessary.
class User:
...
@staticmethod
def __prepare_args__(args: Args) -> Args:
print(f'Preparing arguments for {args.url} request')
return args
Step 3: Make Call¶
When you make the following call
user = get_user(1)
print(user)
user = get_user(1)
print(user)
user = User.get(1)
print(user)
Sensei collects the function's arguments, transforms it to the Args
instance, and calls preparer if it's defined.
Preparing arguments for /users/1 request
After Sensei uses prepared arguments for request and makes it.
User(id=1, email="[email protected]", nickname="john")
Usage¶
There are a few examples of using preparers at different levels.
Route (Routed Method)¶
You can use preparers when some request parameter should be retrieved as the attribute of a Routed Model object from which the method was called
from sensei import APIModel, format_str, Router, Args
from pydantic import NonNegativeInt, EmailStr
router = Router('https://api.example.com')
class User(APIModel):
id: NonNegativeInt
email: EmailStr
nickname: str
@router.patch('/users/{id_}')
def update(
self,
name: str,
job: str
) -> None:
pass
@update.prepare
def _update_in(self, args: Args) -> Args:
args.url = format_str(args.url, {'id_': self.id})
return args
Info
Don't confuse route level (as routed method level) with routed model level. The example above is route level, not routed model level.
Router/Routed Model¶
You can use preparers when all requests require a set of the same parameters, like
Authorization
header. These preparers are executed for each routed function, related to the router
that the preparer is associated with.
Router¶
To apply preparer at the router level, you need to use argument __prepare_args__
of the Router
constructor.
from sensei import APIModel, Router, Args
from pydantic import NonNegativeInt, EmailStr
class User(APIModel):
id: NonNegativeInt
email: EmailStr
nickname: str
class Context:
token: str
def prepare_args(args: Args) -> Args:
args.headers['Authorization'] = f'Bearer {Context.token}' # (1)!
return args
router = Router('https://api.example.com', __prepare_args__=prepare_args)
@router.patch('/users/{id_}')
def update_user(id_: int, nickname: str) -> None:
pass
@router.post('/users')
def create_user(email: EmailStr, nickname: str) -> User:
pass
Context.token = 'secret_token'
id_ = create_user('[email protected]', 'john').id
update_user(id_, 'john_good')
- Context class is used for dynamic retrieving of auth token
Authorization
header will be added to create_user(...)
and update_user(...)
requests.
Routed Model¶
To apply preparer at the routed model level, you need to use hook __prepare_args__
.
from sensei import APIModel, Router, Args, format_str
from pydantic import NonNegativeInt, EmailStr
router = Router('https://api.example.com')
class Context:
token: str
class User(APIModel):
id: NonNegativeInt
email: EmailStr
nickname: str
@staticmethod
def __prepare_args__(args: Args) -> Args:
args.headers['Authorization'] = f'Bearer {Context.token}' # (1)!
return args
@router.patch('/users/{id_}')
def update(self, nickname: str) -> None:
pass
@update.prepare
def _update_in(self, args: Args) -> Args:
args.url = format_str(args.url, {'id_': self.id})
return args
@classmethod
@router.post('/users')
def create(cls, email: EmailStr, nickname: str) -> "User":
pass
Context.token = 'secret_token'
user = User.create('[email protected]', 'john')
user.update('john_good')
- Context class is used for dynamic retrieving of auth token
In the example above, there are two preparers of different levels.
@staticmethod
def __prepare_args__(args: Args) -> Args:
args.headers['Authorization'] = f'Bearer {Context.token}'
return args
@update.prepare
def _update_in(self, args: Args) -> Args:
args.url = format_str(args.url, {'id_': self.id})
return args
According to Order Levels, preparer at routed model level (__prepare_args__
) is executed before
route level preparer (_update_in
).
Tip
If some Router/Routed Model level preparer should be excluded for some routes, you can use skip_preparer=True
in the route
decorator
class User(APIModel):
...
@router.get('/users/{id_}', skip_preparer=True)
def get(self, id_: int) -> "User":
pass
But this doesn't exclude route level preparers, like that:
@get.preparer()
def _get_in(self, args: Args) -> Args:
return args
Finalizers¶
There are two types of finalizers: response finalizer and JSON finalizer. Let's explore them:
Response Finalizer¶
Response Finalizer is a function that takes an instance of httpx.Response
as the argument and returns the result of calling
the associated routed function (method). The return value must be of the same type as the routed function (method).
Response Finalizers are used for response transformation, which can't be performed automatically if you set a corresponding response type from the category of automatically handled.
Info
Response Finalizers can be defined only at the route level.
If response type is not from the category automatically handled, you have to define a response finalizer. Otherwise, an error will be thrown.
Algorithm¶
Look at the example below:
from sensei import Router, APIModel, Form
from pydantic import EmailStr
from typing import Annotated
from httpx import Response
router = Router('https://api.example.com')
class UserCredentials(APIModel):
email: EmailStr
password: str
@router.post('/register')
def sign_up(user: Annotated[UserCredentials, Form(embed=False)]) -> str:
pass
@sign_up.finalize
def _sign_up_out(response: Response) -> str:
print(f'Finalizing response for request {response.request.url}')
return response.json()['token']
token = sign_up(UserCredentials(
email='[email protected]',
password='secret_password')
)
print(f'JWT token: {token}')
Finalizing response for request /register
JWT token: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdW...
Let's take this code step by step.
Step 1: Define Route¶
First, we need to define a route:
@router.post('/register')
def sign_up(user: Annotated[UserCredentials, Form(embed=False)]) -> str:
pass
Step 2: Define Finalizer¶
To define a response finalizer, you have to create a function, that takes only one argument of the httpx.Response
type and returns
a value of the same type as the routed function. After that, decorate it as @<routed_function>.finalize
.
@sign_up.finalize
def _sign_up_out(response: Response) -> str:
print(f'Finalizing response for request {response.request.url}')
return response.json()['token']
Step 3: Make Call¶
When you make the following call
token = sign_up(UserCredentials(
email='[email protected]',
password='secret_password')
)
print(f'JWT token: {token}')
Sensei makes a request, retrieves the response, and passes the Response
object to the finalizer if it's defined.
Finalizing response for request /register
Finally, the routed function returns the result of the finalizer call. In this example, the result is a JWT token.
JWT token: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdW...
Usage¶
Response Finalizers are used for response transformation, that can't be performed automatically, if set a corresponding response type from the category of automatically handled.
Example
This example was shown in Algorithm
from sensei import Router, APIModel, Form
from pydantic import EmailStr
from typing import Annotated
from httpx import Response
router = Router('https://api.example.com')
class UserCredentials(APIModel):
email: EmailStr
password: str
@router.post('/register')
def sign_up(user: Annotated[UserCredentials, Form(embed=False)]) -> str:
pass
@sign_up.finalize
def _sign_up_out(response: Response) -> str:
print(f'Finalizing response for request {response.request.url}')
return response.json()['token']
token = sign_up(UserCredentials(
email='[email protected]',
password='secret_password')
)
print(f'JWT token: {token}')
JSON Finalizer¶
JSON Finalizer is a function that takes the decoded JSON response, as the argument and returns the modified JSON. These finalizers are used for JSON response transformation before internal or user-defined response finalizing.
Info
JSON Finalizers can be defined only at the router/routed model level.
Algorithm¶
For instance, each response wraps primary data in a field "data"
Like that API response:
HTTP/1.1 200 OK
Content-Type: application/json
{
"status": "success",
"data": {
"id": 123,
"name": "Alice",
"email": "[email protected]"
}
}
To automatically handle the response, you either need to define this model:
class _UserData(APIModel):
id: int
name: str
email: EmailStr
class User(APIModel):
data: _UserData
Which is inconvenient to use, or define the next response finalizer
class User(APIModel):
id: int
name: str
email: EmailStr
@router.get('/users/{id_}')
def get_user(id_: Annotated[int, Path(alias='id')]) -> User:
pass
@get_user.finalize
def _user_out(response: Response) -> User:
return User(**response.json()['data'])
The approach with a response finalizer returns a better model than the approach with wrapping model in another model.
But there is a problem, what if you have dozens of similar functions and you have to define the same finalizer for each? JSON finalizer is the solution to this problem.
Let's try to figure out the algorithm for the situation described above:
from sensei import Router, APIModel, Query, Path
from typing import Any, Annotated
from pydantic import EmailStr, PositiveInt, AnyHttpUrl
def _finalize_json(json: dict[str, Any]) -> dict[str, Any]:
return json['data']
router = Router('https://api.example.com', __finalize_json__=_finalize_json)
class User(APIModel):
email: EmailStr
id: PositiveInt
name: str
@router.get('/users')
def query_users(
page: Annotated[int, Query()] = 1,
per_page: Annotated[int, Query(le=7)] = 3
) -> list[User]:
pass
@router.get('/users/{id_}')
def get_user(id_: Annotated[int, Path(alias='id')]) -> User:
pass
user = get_user(1)
print(user)
from sensei import Router, APIModel, Query, Path
from typing import Any, Annotated
from typing_extensions import Self
from pydantic import EmailStr, PositiveInt, AnyHttpUrl
router = Router('https://api.example.com')
class User(APIModel):
email: EmailStr
id: PositiveInt
name: str
@classmethod
def __finalize_json__(cls, json: dict[str, Any]) -> dict[str, Any]:
return json['data']
@classmethod
@router.get('/users')
def query(
cls,
page: Annotated[int, Query()] = 1,
per_page: Annotated[int, Query(le=7)] = 3
) -> list[Self]:
pass
@classmethod
@router.get('/users/{id_}')
def get(cls, id_: Annotated[int, Path(alias='id')]) -> Self:
pass
user = User.get(1)
print(user)
User(email='[email protected]' id=1 name='John')
Let's take this code step by step.
Step 1: Define Route¶
First, we need to define a route(s)
@router.get('/users')
def query_users(
page: Annotated[int, Query()] = 1,
per_page: Annotated[int, Query(le=7)] = 3
) -> list[User]:
pass
@router.get('/users/{id_}')
def get_user(id_: Annotated[int, Path(alias='id')]) -> User:
pass
@classmethod
@router.get('/users')
def query(
cls,
page: Annotated[int, Query()] = 1,
per_page: Annotated[int, Query(le=7)] = 3
) -> list[Self]:
pass
@classmethod
@router.get('/users/{id_}')
def get(cls, id_: Annotated[int, Path(alias='id')]) -> Self:
pass
Step 2: Define Finalizer¶
To define a JSON finalizer, you have to create a function, that takes only one argument of some JSON-serializable type
corresponding to the response. Usually, servers return JSON as dict[str, Any]
, but rarely can return list[dict]
.
The next step is based on a level type:
Pass function to Router
object
def _finalize_json(json: dict[str, Any]) -> dict[str, Any]:
return json['data']
router = Router('https://api.example.com', __finalize_json__=_finalize_json)
You only need to define a class or static method with the name __finalize_json__
inside User
. Sensei will use this hook, when it's necessary.
@classmethod
def __finalize_json__(cls, json: dict[str, Any]) -> dict[str, Any]:
return json['data']
Step 3: Make the call¶
When you make the following call
user = get_user(1)
print(user)
user = User.get(1)
print(user)
Sensei makes a request, retrieves the response, decodes it as JSON, and passes it to the response finalizer if it's defined.
User(email='[email protected]' id=1 name='John')
That is, the JSON finalizer is executed before the response finalizer. Due to this, the Self
response type can be handled
without a response finalizer.
Usage¶
JSON finalizers are used for JSON response transformation before internal or user-defined response finalizing. The example was shown in Algorithm
Tip
If some Router/Routed Model level finalizer should be excluded for some routes, you can use skil_finalizer=True
in the route:
class User(APIModel):
...
@classmethod
@router.get('/users/{id_}', skip_finalizer=True)
def get(cls, id_: Annotated[int, Path(alias='id')]) -> Self:
pass
Async¶
Preparers/Finalizers can be async if the associated routed function is also async.
class User(APIModel):
...
@router.patch('/users/{id_}')
async def update(
name: str,
job: str
) -> datetime.datetime:
pass
@update.prepare
async def _update_in(self, args: Args) -> Args:
args.url = format_str(args.url, {'id_': self.id})
await asyncio.sleep(1.5)
return args
@update.finalize
async def _update_out(self, response: Response) -> datetime.datetime:
json_ = response.json()
result = datetime.datetime.strptime(json_['updated_at'], "%Y-%m-%dT%H:%M:%S.%fZ")
await asyncio.sleep(1.5)
self.first_name = json_['name']
return result
Recap¶
In summary, Sensei provides flexible hooks to customize how requests are prepared and responses are handled through Preparers and Finalizers. These hooks can be applied at different levels with an organized execution order, providing nuanced control over request handling and response processing:
-
Hook Levels
Sensei defines three levels for hooks:
- Routed Model Level (executed first) - Define within the routed model, to use across routed methods
- Router Level (executed next) - Attach to a
Router
instance to use across routes - Route Level (executed last) - Directly to a specific route
Each of these levels allows you to add preparers or finalizers that operate at different scopes, with Routed Model Level being the most specific and executed first.
-
Preparers
Preparers are used to modify or add arguments before a request is sent.
Example: Preparers are commonly used to add authentication headers or to dynamically set route parameters (e.g., filling in path variables from the instance attributes).
-
Finalizers
Finalizers process responses after a request. They come in two forms:
- Response Finalizer: Operates on the
httpx.Response
object, transforming it into the final returned data. - JSON Finalizer: Works on JSON data from the response, modifying it before other processing. JSON finalizers are helpful for removing unnecessary nested structures or normalizing data formats.
Example: If an API response wraps data in an additional "data" field, a JSON finalizer can remove this wrapper automatically for all routes associated with the router or routed model. The Response Finalizer can then transform the processed JSON into the model or data structure defined in the route.
- Response Finalizer: Operates on the