HTTPX Framework¶

Installation:

pip install chttpx # or cli2[httpx]

Then:

import chttpx

The goal of chttpx module is to provide a generic framework to build HTTP client libs and CLIs onto, after carrying this pattern from a project to another, I’ve refactored this stuff here:

great logging: Sent/Received JSON output is dumped as YAML, colored in console, and not-colored to file
export LOG_LEVEL=DEBUG to enable debug output
debug output always saved in ~/.local/cli2/log, they will eventually fill up your drive and I’ve not yet decided a solution against that, but I just love this feature,
absolutely beautiful HTTP Exceptions
export HTTP_DEBUG=1 for low-level HTTP Debugging output
a ORM for REST resources

Example¶

And of course all this is designed to combine very well with CLIs, because once you have a library for an API, which you’re going to embed in god knows what (your API server, an Ansible plugin …), you’ll want to work with a CLI to debug stuff: discover the API and implement features incrementally.

Source code¶

import cli2
import chttpx


class APIClient(chttpx.Client):
    """
    Client for restful-api.dev

    Prior to using this, run at the root of this repository:

    .. code-block::

        pip install django djangorestframework
        ./manage.py migrate
        ./manage.py runserver
    """
    mask_keys = ['Capacity']

    def __init__(self, *args, **kwargs):
        kwargs.setdefault('base_url', 'http://localhost:8000')
        super().__init__(*args, **kwargs)

    @cli2.cmd
    async def fail(self):
        """ Send bogus Form data """
        await self.post('/foo', data=dict(b=2))


class Object(APIClient.Model):
    """
    restful-api.dev objects

    Example:

    chttpx-example-client object create name=chttpx capacity=2TB
    """
    url_list = '/objects/'
    url_detail = '/objects/{self.id}/'

    id = chttpx.Field()
    name = chttpx.Field()
    capacity = chttpx.Field('data/Capacity')
    generation = chttpx.Field('data/Generation')
    price = chttpx.Field('data/Price')

    @cli2.cmd
    @classmethod
    async def fail(cls):
        """ Send bogus JSON """
        await cls.client.post('/foo', json=dict(a=1))

    @cli2.cmd
    async def rename(self, new_name):
        """ Send bogus JSON with an instance"""
        self.name = new_name
        return await self.save()

    async def update(self):
        return await self.client.put(self.url, json=self.data)


cli = APIClient.cli

Mind you, the Object can be used in a Django-ish ORM style and all these CLIs were created with free Sphinx documentation as seen in Example CLI.

Outputs are just beautiful of course:

See a builtin command with a custom command in action:

The debug output is also awesome:

It shows:

the JSON being sent to the server
request/method/url/timestamp
the JSON being returned by the server
response status code returned by the server
finnaly, the return value of the command, which is the created object, see how the returned object was updated with the id and createAt fields which came from the response

Of course, you’re going to be able to override/customize everything as you dig into the API that you’re implementing a client for.

Architecture¶

The client module is built around 3 main moving parts:

Client: A wrapper around the httpx.AsyncClient class,
Handler: Used by the client to automate response handling: do we retry, need to re-create a TCP Session, or get a new token…
Model: A Django-like model metaclass, that comes with it’s Field classes and their expressions

Tutorial¶

Creating a Client¶

Start by extending a Client:

import chttpx

class YourClient(chttpx.Client):
    pass

# you get a CLI for free
cli = YourClient.cli

There are a few methods that you might want to override:

client_factory(): where you can customize the actual httpx AsyncClient instance before it is used by cli2 Client.
token_get(): if you want your client to do some authentication dance to get a token
cli_kwargs: Overrides for the for the cli Group

Pagination¶

The default Paginator doesn’t know how to paginate. Let’s teach it to make a page GET parameter:

class YourClient(chttpx.Client):
    class Paginator(chttpx.Paginator):
        def pagination_parameters(self, params, page_number):
            params['page'] = page_number

That will increments a page GET parameter until it gets an empty results list, which works but is still sub-optimal. Let’s teach it when to stop by setting total_pages in pagination_initialize():

class YourClient(chttpx):
    class Paginator(chttpx.Paginator):
        def pagination_parameters(self, params, page_number):
            params['page'] = page_number

        def pagination_initialize(self, data):
            self.total_pages = data['total_pages']

Perhaps you don’t get the total pages from the API response, but you do get a total number of items, which you can set total_items and total_pages will auto-calculate:

class Paginator(chttpx.Paginator):
    def pagination_initialize(self, data):
        self.total_items = data['total_items']

Perhaps you’re dealing with an offset/limit type of pagination, in which case, page GET parameter won’t do, set offlet/limit instead in pagination_parameters():

class OffsetPagination(chttpx.Paginator):
    def pagination_parameters(self, params, page_number):
        self.per_page = 1
        params['offset'] = (page_number - 1) * self.per_page
        params['limit'] = self.per_page

    def pagination_initialize(self, data):
        self.total_items = data['total']

Creating a Model¶

Then, register a Model for this client by subclassing it’s .Model attribute.

class YourObject(YourClient.Model):
    pass

Several things are happening here:

YourObject._client_class was set to YourClient
YourClient.Models was set to [YourObject]

Now, you’re not supposed to use YourObject directly, but instead get it from the client:

client = YourClient()
model_class = client.YourObject

You can also define a specific paginator:

class YourModel(YourClient.Model):
    paginator = YourPaginator

Model.client¶

As such, the model class you’re using has the client instance set as .client class attribute. And magically, you can use self.client or cls.client anywhere in your model:

class YourObject(YourClient.Model):
    @classmethod
    @cli2.cmd
    async def some_command(cls):
        return await cls.client.get('/some-page').json()

Model.paginate¶

You can already paginate over objects:

async for obj in client.YourObject.paginate('/some-url', somefilter='foo'):
    cli2.print(obj)

If you set the url_list attribute, then you can also use the chttpx.Model.find() method directly:

class YourObject(YourClient.Model):
    pass

paginator = YourObject.find(somefilter='test')

Fields¶

You can also define fields for your Model as such:

class YourModel(YourClient.Model):
    id = chttpx.Field()

You guessed it: this will may the id key of the Model.data to the .id property. Which allows for more interesting things as we’ll see…

Nested fields¶

If you want to map data['company']['name'] to company_name, use slash to define a nested data accessor:

class YourModel(YourClient.Model):
    company_name = chttpx.Field('company/name')

You can also “pythonize” any property with a simple accessor without any slash:

class YourModel(YourClient.Model):
    company_name = chttpx.Field('companyName')

Default factory¶

Default field values can be computed at runtime with the factory() decorator:

class YourModel(YourClient.Model):
    hasdefault = chttpx.Field()

    @hasdefault.factory
    def default_value(self):
        return 'something'

This will cause data to have hasdefault='something'.

If your default value factory depends on other fields, you need to declare them, pass them as argument to factory:

class YourModel(YourClient.Model):
    required1 = chttpx.Field()
    required2 = chttpx.Field()
    hasdefault = chttpx.Field()

    @hasdefault.factory(required1, required2)
    def default_value(self):
        return f'something{self.required1}-{self.required2}'

Custom types¶

The most painful stuff I’ve had to deal with in APIs are datetimes and, “json in json”.

The cures for that are JSONStringField and DateTimeField.

Expressions¶

Sometimes, we want to filter on fields which are not available in GET parameters, in this case, we can filter in Python with SQL-Alchemy-like expressions:

foo_stuff = YourModel.find(YourModel.company_name == 'foo')

You can also pass lambdas:

foo_stuff = YourModel.find(lambda item: item.company_name.lower() == 'foo')

Combine ands and ors:

foo_stuff = YourModel.find(
    (
        # all stuff with company starting with foo
        (lambda item: item.company_name.lower().startswith('foo'))
        # AND ending with bar
        & (lambda item: item.company_name.lower().endswith('bar'))
    )
    # OR with name test
    | YourModel.company_name == 'test'
)

Parameterable¶

Note that we want to delegate as much filtering as we can to the endpoint. To delegate a filter to the endpoint, add a parameter:

class YourModel(YourClient.Model):
    name = chttpx.Field(parameter='name')

This will indicate to the paginator that, given the following expression:

YourModel.find(YourModel.name == 'bar')

The paginator will add the ?name=bar parameter to the URL.

This is nice when you want to just start coding then with only expressions and not bother about which field is parameterable or not.

This looks a bit weak and of limited use as-is, because I haven’t open-sourced the OData part of my code yet, but that is able to generate a query string with nested or/and/startswith/etc. That part won’t end up in the core module anyway, probably a cli2.contrib.odata module.

And I’m sure there are several other more or less protocols out there to do this kind of things, so, we might as well have that here available for free.

Testing¶

chttpx also registers as a pytest plugin, because as you know, I’m pretty lazy when it comes to repetitive test writting which is why I developed django-dbdiff and django-responsediff and also cli2.test.autotest. Let’s have the same thing with chttpx!

Let’s write a test that calls the object create and delete command, say, in the tests/test_client_test.py file:

@pytest.mark.chttpx_mock
def test_object_story():
    test_name = 'test3331'

    obj = APIClient.cli['object']['create'](f'name={test_name}')
    assert obj.name == test_name

    with pytest.raises(chttpx.RefusedResponseError):
        # test_name already exists!
        APIClient.cli['object']['create'](f'name={test_name}')

    APIClient.cli['object']['delete'](f'{obj.id}')

The first time you run this test, our example APIClient will connect to localhost:8000 as it’s configured by default, and actual queries will be exeuted:

[21/Mar/2025 10:35:11] "POST /objects/ HTTP/1.1" 201 38
Bad Request: /objects/
[21/Mar/2025 10:35:11] "POST /objects/ HTTP/1.1" 400 50
[21/Mar/2025 10:35:11] "GET /objects/121/ HTTP/1.1" 200 38
[21/Mar/2025 10:35:11] "DELETE /objects/121/ HTTP/1.1" 204 0

And the chttpx_mock pytest marker will cause new contents to be written for you in tests/fixtures/tests_test_client_test.py::test_object_story.yaml:

- request:
        event: request
        json:
          name: test33312
        level: debug
        method: POST
        timestamp: '2025-03-21 10:38:29'
        url: http://localhost:8000/objects/
  response:
        event: response
        json:
          data: {}
          id: 122
          name: test33312
        level: info
        method: POST
        status_code: '201'
        timestamp: '2025-03-21 10:38:29'
        url: http://localhost:8000/objects/
- request:
        event: request
        json:
          name: test33312
        level: debug
        method: POST
        timestamp: '2025-03-21 10:38:29'
        url: http://localhost:8000/objects/
  response:
        event: response
        json:
          name:
          - object with this name already exists.
        level: info
        method: POST
        status_code: '400'
        timestamp: '2025-03-21 10:38:29'
        url: http://localhost:8000/objects/
# and so on ...

You are supposed to add this file in git, because next time you run the test: the chttpx_mock marker will provision pytest-httpx’s httpx_mock will all the request/responses that have been recorded in the fixture file.

As such, two new pytest options are added by the chttpx plugin:

--chttpx-live: don’t use fixtures at all, run against the real network
--chttpx-rewrite: force rewriting all fixtures

When specifications change, you can remove a given test fixture and run the test again which will rewrite it, or, run with --chttpx-rewrite to rewrite all fixtures.

Danger

Because your fixtures are in git, this will cause a diff in the fixtures file that you will need to review. It’s your responsibility to review these changes properly, we just write the test fixtures for you, but you have to proof-read them!

Patterns¶

In this section, we’ll document various patterns found over time refactoring complex clients.

Customizing Commands¶

You can customize the generated commands in the following methods of the Client class:

setargs(): to set CLI only arguments.
factory(): to construct your Client with the said cli only arguments
post_call(): to logout or do whatever you want

Example:

class CyberArkClient(chttpx.Client):
    def __init__(self, something, *args, **kwargs):
        self.something = something
        super().__init__(*args, **kwargs)

    @classmethod
    def setargs(self, cmd):
        # declare an argument that will be visible in command line only
        cmd.arg('something', position=0, kind='POSITIONAL_ONLY')

    async def factory(cls, something):
        # something will be passed by the ClientCommand class
        return cls(something)

    async def token_get(self):
        # do something to get a token
        return token

    async def post_call(self, cmd):
        # release the token
        await self.client.post('/logoff')

Filtering on external data¶

You may want to be able to filter on fields which won’t be returned by the list API:

class DynatraceConfiguration(YourClient.Model):
    url_list = '/configurations'

    async def status_fetch(self):
        response = self.client.get(self.url + '/status')
        self.status = response.json()['status']

    @classmethod
    @cli2.cmd
    async def find(cls, *expressions, **params):
        paginator = super().find(
            lambda item: item.status == 'OK',
            *expressions,
            **params,
        )

        async def callback(item):
            await item.status_fetch()

        paginator.callback = callback
        return paginator

Before yielding items, paginator will call the callback for every item in asyncio.gather, causing an extra async request to the status URL of the object and set self.status, this will cause a lot of requests, you might want to set semaphore to limit concurrent requests.

API¶

HTTP Client boilerplate code to conquer the world.

class chttpx.Client(*args, handler=None, semaphore=None, mask=None, debug=False, **kwargs)[source]¶

HTTPx Client Wrapper

paginator¶: Paginator class

semaphore¶: Optionnal asyncio semaphore to throttle requests.

handler¶: A callback that will take responses objects and decide wether or not to retry the request, or raise an exception, or return the request. Default is a Handler

mask_keys¶

Use this class attribute to declare keys to mask:

class YourClient(cli2.Client):
    mask_keys = ['password', 'secret']

cli¶: Generated Group for this client. Uses cli_kwargs to pass kwargs to the generated group. Note that this is a cached property.

cli_kwargs¶

Dict of overrides for the generated Group. Example:

class YourClient(cli2.Client):
    cli_kwargs = dict(cmdclass=YourCommandClass)

cmdclass¶: ClientCommand class or subclass. You usually won’t have to define this, instead, you should do what you need in the factory(), setargs() and post_call() methods.

debug¶: Enforce full logging: quiet requests are logged, masking does not apply. This is also enabled with environment variable DEBUG.

mask¶: Mask object

models¶: Declared models for this Client.

class Model(data=None, **values)¶

cmdclass¶: alias of ModelCommand

property client¶: Return last client object used, unless it raised RemoteProtocolError.

client_factory()[source]¶: Return a fresh httpx async client instance.

client_token_apply(client)[source]¶

Actually provision self.client with self.token.

This is yours to implement, ie.:

client.headers['X-API'] = f'Bearer {self.token}'

Do NOT use self.client in this function given it’s called by the factory itself.

Parameters:: client – The actual AsyncClient instance to provision.

cmdclass¶: alias of ClientCommand

async delete(url, *args, **kwargs)[source]¶: DELETE Request

async classmethod factory()[source]¶

Override this method to customize your client construction.

You can add custom args, if you declare them in setargs().

async get(url, *args, **kwargs)[source]¶: GET Request

async head(url, *args, **kwargs)[source]¶: HEAD Request

paginate(url, *expressions, params=None, model=None, callback=None)[source]¶

Return a paginator to iterate over results

Parameters:

url – URL to paginate on
params – GET parameters
model – Model class to cast for items

paginator¶: alias of Paginator

async patch(url, *args, **kwargs)[source]¶: PATCH Request

async post(url, *args, **kwargs)[source]¶: POST Request

async post_call(cmd)[source]¶

Override this method which will run after a CLI exits.

Parameters:: cmd – ClientCommand object

async put(url, *args, **kwargs)[source]¶: PUT Request

async request(method, url, *, handler=None, quiet=False, accepts=None, refuses=None, tries=None, backoff=None, retries=True, semaphore=None, mask=None, content=None, data=None, files=None, json=None, params=None, headers=None, cookies=None, auth=<httpx._client.UseClientDefault object>, follow_redirects=<httpx._client.UseClientDefault object>, timeout=<httpx._client.UseClientDefault object>, extensions=None)[source]¶

Request method

If your client defines a token_get callable, then it will automatically play it.

If your client defines an asyncio semaphore, it will respect it.

client = Client()

await client.request(
    'GET',
    '/',
    # extend the current handler with 10 tries with 200 accepted
    # status code only
    tries=10,
    accepts=[200],
)

await client.request(
    'GET',
    '/',
    # you can also pass your own handler callable
    handler=Handler(tries=10, accepts=[200]),
    # that could also have been a function
)

Parameters:

method – HTTP Method name, GET, POST, etc
url – URL to query
handler – If a callable, will be called, if a dict, will extend the client’s handler.
quiet – Wether to log or not, used by Paginator to not clutter logs with pagination. Meaning if you want to debug pagination, you’ll have to make it not quiet from there. If you really want to see all results, set debug to True.
retries – Wether to retry or not in case handler dosen’t accept the response, set to False if you want only 1 try.
accepts – Override for Handler.accepts
refuses – Override for Handler.refuses
tries – Override for Handler.tries
backoff – Override for Handler.backoff
semaphore – Override for Client.semaphore

async request_cmd(method, url, *args, **kwargs)[source]¶

Perform an HTTP Request.

This calls the underlying httpx.Client request command, so, you can use kwargs such as content for raw body pass, data for form data, and json for json. Values for these kwargs may be file paths.

Example:

request POST /objects json=my_data.yaml

Parameters:

method – HTTP verb, GET, POST, etc
url – URL relative to the client’s base_url
args – Any args to pass to the request method
kwargs – Any kwargs that will be loaded as file

async send(request, handler, retries=True, semaphore=None, log=None, auth=None, follow_redirects=None)[source]¶: Internal request method

classmethod setargs(cmd)[source]¶

Override this method to declare CLI args globally for this client.

Parameters:: cmd – ClientCommand object

async token_get()[source]¶

Authentication dance to get a token.

This method will automatically be called by request() if it finds out that token is None.

This is going to depend on the API you’re going to consume, basically it’s very client-specific.

By default, this method does nothing. Implement it to your likings.

This method is supposed to return the token, but doesn’t do anything with it by itself.

You also need to implement the client_token_apply() which is in charge of updating the actual httpx client object with the said token.

async def token_get(self):
    response = await self.post('/login', dict(...))
    return response.json()['token']

def client_token_apply(self, client):
    client.headers['X-ApiKey'] = self.token

async token_refresh()[source]¶: Use token_get() to get a token

class chttpx.ClientCommand(target, *args, **kwargs)[source]¶

Client CLI command

client¶: The client object that was constructed from Client.factory()

async factories_resolve()[source]¶: Set client after resolving factories.

async post_call()[source]¶: Call Client.post_call().

setargs()[source]¶: Set a self factory of Client.factory() method, and call Client.setargs().

exception chttpx.ClientError[source]¶

class chttpx.DateTimeField(*args, fmt=None, fmts=None, **kwargs)[source]¶

JSON has no datetime object, which means different APIs will serve us different formats in string variables.

Heck, I’m pretty sure there are even some APIs which use different formats. This is the cure the world needed against that disease.

fmt¶: The datetime format for Python’s strptime/strftime.

fmts¶: A list of formats, in case you don’t have one. This list will be populated with default_fmts by default.

default_fmts¶: A class property containing a list of formats we’re going to try to figure fmt and have this thing “work by default”. Please contribute to this list with different formats.

externalize(obj, value)[source]¶: Convert the internal string into an external datetime.

internalize(obj, value)[source]¶: Convert a datetime into an internal string.

class chttpx.Field(data_accessor=None, parameter=None)[source]¶

Field descriptor for models.

The basic Field descriptor manages (get and set) data from within the Model.data JSON.

Since sub-classes are going to convert data, we need to understand the concepts of internal and external data:

external: the Python value, this can be any kind of Python object
internal: the Python representation of the JSON value, this can be any Python type that will work given json.dumps
internalizing: process of converting a Python value into a JSON one
externalizing: process of converting a JSON value into something Python

data_accessor¶: Describes how to access the Field’s data inside the model’s data dict. If data_accessor is foo then this field will control the foo key in the model’s data dict. Use / to separate keys when nesting, if data_accessor is foo/bar then it will control the bar key of the foo dict in the model’s data dict.

parameter¶: Name of the GET parameter on the model’s Model.url_list, if any. So that the filter will be converted to a GET parameter. Otherwise, filtering will happen in Python.

clean(obj)[source]¶

Clean the data.

Called by the Model when requested a data, this method:

Gets the externalized value from __get__()
Convert it into a JSON object with internalize()
Uses internal_set() to update Model.data

externalize(obj, value)[source]¶

Transform internal JSON value to Python value, based on data_accessor.

Any kind of processing from the JSON value to the Python value can be done in this method.

internal_get(obj)[source]¶: Get the “raw” value from the object, which is a Python representation of the JSON internal value, using data_accessor.

internal_set(obj, value)[source]¶: Using data_accessor, set the value in Model.data

internalize(obj, value)[source]¶

Transform external value into JSON internal value.

Any kind of processing from the Python value to the JSON value can be done in this method.

mark_dirty(obj)[source]¶: Tell the model that the data must be cleaned.

exception chttpx.FieldError[source]¶

exception chttpx.FieldExternalizeError(msg, field, obj, value)[source]¶

exception chttpx.FieldValueError(msg, field, obj, value)[source]¶

class chttpx.Handler(accepts=None, refuses=None, retokens=None, tries=None, backoff=None)[source]¶

tries¶: Number of retries for an un-accepted request prior to failing. Default: 30

backoff¶: Will sleep number_of_tries * backoff prior to retrying. Default: .1

accepts¶: Accepted status codes, you should always set this to ensure responses with an unexpected status either retry or raise. Default: range(200, 299)

refuses¶: List of refused status codes, responses returning those will not retry at all and raise directly. Default: [400, 404]

retokens¶: Status codes which trigger a new call of token_get() prior to a retry. Only one retry is done then by this handler, considering that authenticating twice in a row is useless: there’s a problem in your credentials instead. Default: [401, 403, 407, 511]

class chttpx.JSONStringField(*args, options=None, **kwargs)[source]¶

Yes, some proprietary APIs have JSON fields containing JSON strings.

This Field is the cure the world needed for that disease.

options¶: Options dict for json.dumps, ie. options=dict(indent=4)

externalize(obj, value)[source]¶

Transform internal JSON value to Python value, based on data_accessor.

Any kind of processing from the JSON value to the Python value can be done in this method.

internalize(obj, data)[source]¶

Transform external value into JSON internal value.

Any kind of processing from the Python value to the JSON value can be done in this method.

class chttpx.Model(data=None, **values)[source]¶

You should never call this class directly, instead, get it from the Client object after decorating your model classes with a client as such:

paginator¶: Paginator class, you can leave it by default and just implement pagination_initialize() and pagination_parameters().

url_list¶: The URL to get the list of objects, you’re supposed to configure it as a model attribute in your model subclass. This may be a format string using a client client variable.

url_detail¶: The URL to get the details of an object, you’re supposed to configure it as a model attribute in your model subclass.

id_field¶: Name of the field that should be used as resource identifier, id by default.

url¶: Object URL based on url_detail and id_field.

cli_kwargs¶: Dict of kwargs to use to create the Group for this model.

cmdclass¶: ModelCommand subclass. You generally don’t need to define this, instead, you should do what you need in the Client.factory(), Client.setargs() and Client.post_call() methods.

cmdclass¶: alias of ModelCommand

async classmethod create(**kwargs)[source]¶: Instanciate a model with kwargs and run save().

property data¶: Just ensure we update dirty data prior to returning the data dict.

async delete()[source]¶

Delete model.

DELETE request on url

classmethod find(*expressions, **params)[source]¶

Find objects filtered by GET params

Parameters:

params – GET URL parameters
expressions – Expression list

async classmethod get(**kwargs)[source]¶: Instanciate a model with kwargs and run hydrate().

async hydrate(data=None)[source]¶

Refresh data with GET requset on url_detail

Parameters:: data – Data dict, otherwise will get it

property id_value¶: Return value of the id_field.

async instanciate()[source]¶

POST data to url_list, update data with response json.

You might want to override this.

classmethod paginate(url, *expressions, **params)[source]¶: Return a Paginator based on url_list :param expressions: Expression list

async save()[source]¶

Call update() if self.id otherwise instanciate().

Then updates data based on the response.json if possible.

You might want to override this.

async update()[source]¶

POST data to url_list, update data with response json.

You might want to override this.

class chttpx.ModelCommand(target, *args, **kwargs)[source]¶

Command class for Model class.

async factories_resolve()[source]¶: Return a client object from it’s factory, will all args resolved.

async get_model()[source]¶: Return a client instance bound model

async post_call()[source]¶: Implement your cleaner here

setargs()[source]¶: ModelCommand setargs which calls setargs on the client class and defines an id argument for object commands.

class chttpx.Paginator(client, url, params=None, model=None, expressions=None, callback=None)[source]¶

Generic pagination class.

Should work with most paginations by default, if you’re extending this then override:

pagination_initialize()
pagination_parameters()

total_pages¶: Total number of pages.

total_items¶: Total number of items.

per_page¶: Number of items per page

url¶: The URL to query

url¶: The URL to query

params¶: Dictionnary of GET parameters

model¶: Model class or dict by default.

callback¶: Async callback called for every item before filtering by expressions.

async call(callback)[source]¶

Call an async callback for each item

Parameters:: callback – Function to call for every item.

data_items(data)[source]¶

Given response data, return items.

Parameters:: data – Response JSON data

async first()[source]¶: Return first item

async initialize(response=None)[source]¶

This method is called once when we get the first response.

Parameters:: response – First response object

async last_item()[source]¶: Return the last item of a paginated request.

async list()[source]¶: Return casted list of items

async page_items(page_number)[source]¶

Return the items of a given page number.

Parameters:: page_number – Page number to get the items from

async page_response(page_number)[source]¶

Return the response for a page.

Parameters:: page_number – Page number to get the items from

pagination_initialize(data)[source]¶

Initialize paginator based on the data of the first response.

If at least, you can set total_items or total_pages, per_page would also be nice.

Parameters:: data – Data of the first response

pagination_parameters(params, page_number)[source]¶

Return GET parameters for a given page.

Calls Model.pagination_parameters() if possible otherwise Client.pagination_parameters().

You should implement something like this in your model or client to enable pagination:

def pagination_parameters(self, params, page_number):
    params['page'] = page_number

Parameters:

params – Dict of base GET parameters
page_number – Page number to get

response_items(response)[source]¶

Parse a response and return a list of model items.

Parameters:: response – Response to parse

reverse()[source]¶

Return a copy of this Paginator object to iterate in reverse order.

For this to work, pagination_initialize() must set per_page and either of total_pages or total_items, which is up to you to implement.

exception chttpx.RefusedResponseError(client, response, tries, msg=None)[source]¶

class chttpx.Related(model, many=False, *args, **kwargs)[source]¶

Related model field.

model¶: STRING name of the related model class.

many¶: Set this to True if you’re expecting a list of models in the field.

externalize(obj, value)[source]¶: Instanciate the related model class with the value.

internalize(obj, data)[source]¶: Return the related object’s data.

exception chttpx.ResponseError(client, response, tries, msg=None)[source]¶

Beautiful Response Error class.

response¶: httpx Response object

request¶: httpx Request object

status_code¶: Response status code

url¶: Request url

method¶: Request method

enhance(msg)[source]¶

Enhance an httpx.HTTPStatusError

Adds beatiful request/response data to the exception.

Parameters:: exc – httpx.HTTPStatusError

exception chttpx.RetriesExceededError(client, response, tries, msg=None)[source]¶

exception chttpx.TokenGetError(client, response, tries, msg=None)[source]¶

Example CLI¶

chttpx-example¶

chttpx-example ¶

Client for restful-api.dev

Prior to using this, run at the root of this repository:

pip install django djangorestframework
./manage.py migrate
./manage.py runserver

Sub-Command	Help
`help`	Get help for a command or group
`get`	GET Request
`request`	Perform an HTTP Request
`fail`	Send bogus Form data
`object`	restful-api.dev objects

chttpx-example get¶

chttpx-example get URL [ARGS]... [KWARGS=VALUE]... ¶

Function: chttpx.Client.get()

GET Request

Argument	Help
`URL`	Required: True
`[ARGS]...`	Required: True usage: Any un-named arguments, ie.: `something other`
`[KWARGS=VALUE]...`	Required: True Usage: Any number of named self.arguments, ie.: `something=somevalue other=foo`

chttpx-example request¶

chttpx-example request METHOD URL [ARGS]... [KWARGS=VALUE]... ¶

Function: chttpx.Client.request_cmd()

Perform an HTTP Request.

This calls the underlying httpx.Client request command, so, you can use kwargs such as content for raw body pass, data for form data, and json for json. Values for these kwargs may be file paths.

Example:

request POST /objects json=my_data.yaml

Argument	Help
`METHOD`	HTTP verb, GET, POST, etc Required: True
`URL`	URL relative to the client’s base_url Required: True
`[ARGS]...`	Any args to pass to the request method Required: True usage: Any un-named arguments, ie.: `something other`
`[KWARGS=VALUE]...`	Any kwargs that will be loaded as file Required: True Usage: Any number of named self.arguments, ie.: `something=somevalue other=foo`

chttpx-example fail¶

chttpx-example fail ¶

Function: chttpx.example.APIClient.fail()

Send bogus Form data

chttpx-example object¶

chttpx-example object ¶

restful-api.dev objects

Example:

chttpx-example-client object create name=chttpx capacity=2TB

Sub-Command	Help
`help`	Get help for a command or group
`find`	Find objects filtered by GET params
`delete`	Delete model
`create`	POST request to create
`get`	Get a model based on kwargs
`fail`	Send bogus JSON
`rename`	Send bogus JSON with an instance

chttpx-example object find¶

chttpx-example object find [EXPRESSIONS]... [PARAMS=VALUE]... ¶

Function: chttpx.Model.find()

Find objects filtered by GET params

Argument	Help
`[EXPRESSIONS]...`	`Expression` list Required: True usage: Any un-named arguments, ie.: `something other`
`[PARAMS=VALUE]...`	GET URL parameters Required: True Usage: Any number of named self.arguments, ie.: `something=somevalue other=foo`

Argument

Help

[EXPRESSIONS]...

Expression list

Required: True
usage: Any un-named arguments, ie.: something other

[PARAMS=VALUE]...

GET URL parameters

Required: True
Usage: Any number of named self.arguments, ie.: something=somevalue other=foo

chttpx-example object delete¶

chttpx-example object delete ID ¶

Function: chttpx.Model.delete()

Delete model.

DELETE request on url

Argument	Help
`ID`	ID Required: True

chttpx-example object create¶

chttpx-example object create [KWARGS=VALUE]... ¶

Function: chttpx.Model.create()

Instanciate a model with kwargs and run save().

Argument	Help
`[KWARGS=VALUE]...`	Required: True Usage: Any number of named self.arguments, ie.: `something=somevalue other=foo`

chttpx-example object get¶

chttpx-example object get [KWARGS=VALUE]... ¶

Function: chttpx.Model.get()

Instanciate a model with kwargs and run hydrate().

Argument	Help
`[KWARGS=VALUE]...`	Required: True Usage: Any number of named self.arguments, ie.: `something=somevalue other=foo`

chttpx-example object fail¶

chttpx-example object fail ¶

Function: chttpx.example.Object.fail()

Send bogus JSON

chttpx-example object rename¶

chttpx-example object rename ID NEW_NAME ¶

Function: chttpx.example.Object.rename()

Send bogus JSON with an instance

Argument

Help

ID

Required: True

NEW_NAME

Required: True

HTTPX Framework¶

Example¶

Source code¶

Architecture¶

Tutorial¶

Creating a Client¶

Pagination¶

Creating a Model¶

Model.client¶

Model.paginate¶

Fields¶

Nested fields¶

Default factory¶

Custom types¶

Expressions¶

Parameterable¶

Related¶

Many Related¶

Testing¶

Patterns¶

Customizing Commands¶

Filtering on external data¶

API¶

Example CLI¶

chttpx-example¶

chttpx-example get¶

chttpx-example request¶

chttpx-example fail¶

chttpx-example object¶

chttpx-example object find¶

chttpx-example object delete¶

chttpx-example object create¶

chttpx-example object get¶

chttpx-example object fail¶

chttpx-example object rename¶