Dequest is a full featured declarative HTTP client for Python that simplifies the creation of HTTP requests and retrieves the results as DTO. Here is the package's features:
β Supports GET, POST, PUT, PATCH and DELETE requests
β Sync & Async Client
β Optional Caching for GET Requests (Support In-Memory, Redis, Django Cache)
β Authentication (Static & Dynamic)
β Maps API Json/XML response to DTO object and list (Supports unlimited nested DTOs)
β Support query parameters, JSON body and Form-data
β Implements Retry & Timeout Handling
β Allows Custom Headers per Request (Static & Dynamic)
β Circuit Breaker with Custom Fallback Function
β API parameter mapping and type checking
β Logging
To install Dequest, simply run:
pip install dequest
Dequest allows global configuration via DequestConfig
, the configuration can be set using .config
method of the DequestConfig
class:
from dequest import DequestConfig
DequestConfig.config(
cache_provider="redis", # defaults to "in_memory"
redis_host="my-redis-server.com",
redis_port=6380,
redis_db=1,
redis_password="securepassword",
redis_ssl=True,
)
Use @sync_client
to make synchronous HTTP requests without writing boilerplate code:
from dequest import sync_client, QueryParameter
from typing import List
from dataclasses import dataclass
@dataclass
class UserDto:
id: int
name: str
city: str
def __init__(self, id, name, city):
self.id = id
self.name = name
self.city = city
@sync_client(url="https://jsonplaceholder.typicode.com/users", dto_class=UserDto)
def get_users(city: QueryParameter[str, "city_name"]) -> List[UserDto]:
pass
users = get_users(city="New York")
print(users)
Use @async_client
to make non-blocking HTTP requests:
from dequest import async_client, HTTPMethod
async def callback_function(response):
print(response)
@async_client(url="https://api.example.com/notify", method=HTTPMethod.POST, callback=callback_function)
def notify():
pass
notify()
Pass values inside the URL using PathParameter
:
from dequest import sync_client, PathParameter
@sync_client(url="https://jsonplaceholder.typicode.com/users/{user_id}", dto_class=UserDto)
def get_user(user_id: PathParameter[int]) -> UserDto:
pass
user = get_user(user_id=1)
print(user)
Pass values as URL query parameters using QueryParameter
:
from dequest import sync_client, QueryParameter
@sync_client(url="https://api.example.com/search", dto_class=UserDto)
def search_users(name: QueryParameter[str]):
pass
users = search_users(name="Alice")
For POST requests pass values as JSON payload using JsonBody
:
from dequest import sync_client, HTTPMethod, JsonBody
@sync_client(url="https://api.example.com/users", method=HTTPMethod.POST, dto_class=UserDto)
def create_user(name: JsonBody, city: JsonBody) -> UserDto:
pass
new_user = create_user(name="Alice", city="Berlin")
Automatically retry failed requests:
@sync_client(url="https://api.example.com/data", retries=3, retry_delay=2.0)
def get_data():
pass
Enable caching for GET requests:
@sync_client(url="https://api.example.com/popular-posts", enable_cache=True, cache_ttl=60)
def get_popular_posts():
pass
Prevent excessive calls to failing APIs using a circuit breaker:
from dequest import sync_client, CircuitBreaker
breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=30)
@sync_client(url="https://api.unstable.com/data", circuit_breaker=breaker)
def get_unstable_data():
pass
Define a fallback function for when the circuit breaker is open:
from dequest import CircuitBreaker
def fallback_response():
return {"message": "Service unavailable, returning cached data"}
breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=10, fallback_function=fallback_response)
@sync_client(url="https://api.unstable.com/data", circuit_breaker=breaker)
def fetch_unstable_data():
pass
Dequest is released under the BSD 3-Clause License.