Skip to content

Creating your first consumer

After you've installed DeclarativeX, you can start creating your first consumer.

A consumer is a class that contains methods to interact with a specific API endpoint.

To create a consumer, you need to create a class that inherits from declarativex.Consumer.

from declarativex import Consumer

class MyConsumer(Consumer):
    pass

Your consumer class will have such class methods as:

  • basic_sync - this method will create an httpx client instance for you and pass it to the __init__ method of your consumer.
  • basic_async - this method will create an httpx async client instance for you and pass it to the __init__ method of your consumer.

Both methods accept the same parameters. The only difference is that basic_sync is creating a httpx.Client, while basic_async is creating an httpx.AsyncClient.

Parameter Type Required? Description
base_url str ✅ The base URL of the API you want to consume.
params dict[str, str] ❌ The query parameters you want to pass to the API.
headers dict[str, str] ❌ The headers you want to pass to the API.
follow_redirects bool ❌ Whether to follow redirects.
timeout float ❌ The timeout for the request in seconds.
cookies dict[str, str] ❌ The cookies you want to pass to the API.
auth httpx.BasicAuth | httpx.DigestAuth ❌ The authentication you want to pass to the API.
remove_user_agent bool ❌ Whether to remove the User-Agent header from the request.

You can also use the preconfigured httpx.Client or httpx.AsyncClient instances by passing them directly to the __init__ method of your consumer.

Basic Usage

Here are basic examples of how to initialize a consumer.

consumer = MyConsumer.basic_sync(
    base_url="https://api.example.com",
    params={"key": "value"},
    headers={"Authorization": "Bearer 1234567890"},
    follow_redirects=True,
    timeout=10.0,
    cookies={"session_id": "1234567890"},
    auth=httpx.BasicAuth("username", "password"),
    remove_user_agent=True,
)

All declared endpoints in this consumer will be synchronous.

consumer = MyConsumer.basic_async(
    base_url="https://api.example.com",
    params={"key": "value"},
    headers={"Authorization": "Bearer 1234567890"},
    follow_redirects=True,
    timeout=10.0,
    cookies={"session_id": "1234567890"},
    auth=httpx.BasicAuth("username", "password"),
    remove_user_agent=True,
)

All declared endpoints in this consumer will be asynchronous.

consumer = MyConsumer(
    client=httpx.Client(
        base_url="https://api.example.com",
        params={"key": "value"},
        headers={"Authorization": "Bearer 1234567890"},
        follow_redirects=True,
        timeout=10.0,
        cookies={"session_id": "1234567890"},
        auth=httpx.BasicAuth("username", "password"),
    ),
    remove_user_agent=True,
)

Declared endpoints in this specific example will be synchronous.

Use httpx.AsyncClient if you want to declare asynchronous endpoints.