- Notifications
You must be signed in to change notification settings - Fork2
Meatie is a Python metaprogramming library that eliminates the need for boilerplate code when integrating with REST APIs. Meatie abstracts away mechanics related to HTTP communication, such as building URLs, encoding query parameters, parsing, and dumping Pydantic models.
License
pmateusz/meatie
Folders and files
| Name | Name | Last commit message | Last commit date | |
|---|---|---|---|---|
Repository files navigation
Meatie is a Python library that simplifies the implementation of REST API clients. The library generates code forcalling REST endpoints based on method signatures annotated with type hints. Meatie takes care of mechanics related toHTTP communication, such as building URLs, encoding query parameters, and serializing the body in the HTTP requests andresponses. Rate limiting, retries, and caching are available with some modest extra setup.
Meatie works with all major HTTP client libraries (request, httpx, aiohttp) and offers seamless integration withPydantic (v1 and v2). The minimum officially supported version is Python 3.9.
Generate HTTP clients using type annotations.
fromtypingimportAnnotatedfromaiohttpimportClientSessionfrommeatieimportapi_ref,endpointfrommeatie_aiohttpimportClientfrompydanticimportBaseModel,FieldclassTodo(BaseModel):user_id:int=Field(alias="userId")id:inttitle:strcompleted:boolclassJsonPlaceholderClient(Client):def__init__(self)->None:super().__init__(ClientSession(base_url="https://jsonplaceholder.typicode.com"))@endpoint("/todos")asyncdefget_todos(self,user_id:Annotated[int,api_ref("userId")]=None)->list[Todo]: ...@endpoint("/users/{user_id}/todos")asyncdefget_todos_by_user(self,user_id:int)->list[Todo]: ...@endpoint("/todos")asyncdefpost_todo(self,todo:Annotated[Todo,api_ref("body")])->Todo: ...
Do you use a different HTTP client library in your project? See the example adapted forrequests andhttpx.
https://meatie.readthedocs.io/
Meatie is available onpypi. You can install it with:
pip install meatie
Cache result for a given TTL.
fromtypingimportAnnotatedfrommeatieimportMINUTE,api_ref,cache,endpoint,Cachefrommeatie_aiohttpimportClientfrompydanticimportBaseModelclassTodo(BaseModel): ...classJsonPlaceholderClient(Client):def__init__(self)->None:super().__init__(ClientSession(base_url="https://jsonplaceholder.typicode.com"),local_cache=Cache(max_size=100), )@endpoint("/todos",cache(ttl=MINUTE,shared=False))asyncdefget_todos(self,user_id:Annotated[int,api_ref("userId")]=None)->list[Todo]: ...
A cache key is built based on the URL path and query parameters. It does not include the scheme or the network location.By default, every HTTP client instance has an independent cache. The behavior can be changed in the endpoint definitionto share cached results across all HTTP client class instances.
You can pass your custom cache to the local_cache parameter. The built-in cache provides a max_size parameter to limitits size.
Meatie can delay HTTP requests that exceed the predefined rate limit.
fromtypingimportAnnotatedfromaiohttpimportClientSessionfrommeatieimportLimiter,Rate,api_ref,endpoint,limitfrommeatie_aiohttpimportClientfrompydanticimportBaseModelclassTodo(BaseModel): ...classJsonPlaceholderClient(Client):def__init__(self)->None:super().__init__(ClientSession(base_url="https://jsonplaceholder.typicode.com"),limiter=Limiter(Rate(tokens_per_sec=10),capacity=10), )@endpoint("/todos",limit(tokens=2))asyncdefget_todos(self,user_id:Annotated[int,api_ref("userId")]=None)->list[Todo]: ...
Meatie can retry failed HTTP requests following the strategy set in the endpoint definition. The retry strategy iscontrolled using third-party functions that specify when a retry should be attempted, how long to wait betweenconsecutive attempts to call the endpoint, and whether to abort further retries.
fromtypingimportAnnotatedfromaiohttpimportClientSessionfrommeatieimport (HttpStatusError,RetryContext,after_attempt,api_ref,endpoint,fixed,jit,retry,)frommeatie_aiohttpimportClientfrompydanticimportBaseModelclassTodo(BaseModel): ...defshould_retry(ctx:RetryContext)->bool:ifisinstance(ctx.error,HttpStatusError):returnctx.error.response.status>=500returnFalseclassJsonPlaceholderClient(Client):def__init__(self)->None:super().__init__(ClientSession(base_url="https://jsonplaceholder.typicode.com",raise_for_status=True) )@endpoint("/todos",retry(on=should_retry,stop=after_attempt(3),wait=fixed(5)+jit(2)))asyncdefget_todos(self,user_id:Annotated[int,api_ref("userId")]=None)->list[Todo]: ...
Meatie comes with a built-in set of predefined functions for building retry strategies. Seethemeatie.retry option for more details.
Meatie can inject additional information into the HTTP request. A typical example is adding theAuthorization headerwith a token or signing the request using API keys.
fromtypingimportAnnotated,overridefromaiohttpimportClientSessionfrommeatieimportRequest,api_ref,endpoint,privatefrommeatie_aiohttpimportClientfrompydanticimportBaseModelclassTodo(BaseModel): ...classJsonPlaceholderClient(Client):def__init__(self)->None:super().__init__(ClientSession(base_url="https://jsonplaceholder.typicode.com"))@endpoint("/todos",private)asyncdefget_todos(self,user_id:Annotated[int,api_ref("userId")]=None)->list[Todo]: ...@overrideasyncdefauthenticate(self,request:Request)->None:request.headers["Authorization"]="Bearer bWVhdGll"
Need more control over processing the HTTP requests or responses? See theMeatie Cookbook withsolutions to the most frequently asked questions by the community.
About
Meatie is a Python metaprogramming library that eliminates the need for boilerplate code when integrating with REST APIs. Meatie abstracts away mechanics related to HTTP communication, such as building URLs, encoding query parameters, parsing, and dumping Pydantic models.