Tutorial for cli2.Client: HTTP Client framework¶

Experimental feature, to enjoy it fully, install cli2 with client as such:

pip install cli2[client]

The goal of cli2.client 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


class APIClient(cli2.Client):
    """
    Client for restful-api.dev
    """
    def __init__(self, *args, **kwargs):
        kwargs.setdefault('base_url', 'https://api.restful-api.dev/')
        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:

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

    id = cli2.Field()
    name = cli2.Field()
    capacity = cli2.Field('data/Capacity')
    generation = cli2.Field('data/Generation')
    price = cli2.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 cli2

class YourClient(cli2.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
pagination_initialize(): this is supposed to parse the first response in a paginated query and setup attributes like total_pages
pagination_parameters(): if the endpoint dosen’t support a page GET parameter

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

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 anywhere in your model:

class YourObject(YourClient.Model):
    @classmethod
    async def some_command(cls):
        return await self.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 cli2.client.Model.find() method directly:

class YourObject(YourClient.Model):
    pass

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

You can also customize pagination per-model, in the same fashion as we already can per-client, by implementing and pagination_initialize(), pagination_parameters() in your Model class.

Fields¶

You can also define fields for your Model as such:

class YourModel(YourClient.Model):
    id = cli2.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 = cli2.Field('company/name')

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

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

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
    | item.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 Field.parameter:

class YourModel(YourClient.Model):
    name = cli2.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.

API¶

HTTP Client boilerplate code to conquer the world.

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

HTTPx Client Wrapper

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

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¶: List of keys to mask in logging, ie.: ['password', 'secret']

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

models¶: Declared models for this Client.

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

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.

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

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

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

mask_content(content, mask=None)[source]¶: Implement content masking for non JSON content here.

mask_data(data, mask=None)[source]¶: Apply mask for all mask in data.

paginate(url, params=None, model=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

pagination_initialize(paginator, data)[source]¶

Implement Model-specific pagination initialization here.

Refer to Paginator.pagination_initialize() for details.

pagination_parameters(paginator, page_number)[source]¶

Implement Model-specific pagination parameters here.

Refer to Paginator.pagination_parameters() for details.

paginator¶: alias of Paginator

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

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
mask – Override for Client.mask

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, mask, retries=True, semaphore=None, log=None, auth=None, follow_redirects=None)[source]¶: Internal request method

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

exception cli2.client.ClientError[source]¶

class cli2.client.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 cli2.client.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 cli2.client.FieldError[source]¶

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

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

class cli2.client.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 cli2.client.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 cli2.client.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.

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()[source]¶: Refresh data with GET requset on url_detail

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.

model_command¶: alias of ModelCommand

model_group¶: alias of ModelGroup

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

classmethod pagination_initialize(paginator, data)[source]¶

Implement Model-specific pagination initialization here.

Otherwise, Client.pagination_initialize() will take place.

Refer to Paginator.pagination_initialize() for details.

classmethod pagination_parameters(paginator, page_number)[source]¶

Implement Model-specific pagination parameters here.

Otherwise, Client.pagination_parameters() will take place.

Refer to Paginator.pagination_parameters() for details.

paginator¶: alias of Paginator

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 cli2.client.ModelCommand(target, *args, **kwargs)[source]¶

Command class for Model class.

setargs()[source]¶: Reset arguments.

class cli2.client.ModelGroup(cls)[source]¶

class cli2.client.MutableField(data_accessor=None, parameter=None)[source]¶

Base class for mutable value fields like JSONStringField

Basically, this field:

caches the externalized value, so that you can mutate it
marks the field as dirty so you get the internalized mutated value that next time you get the Model.data

cache_get(obj)[source]¶

Return cached value for obj

Parameters:: obj – Model object

cache_set(obj, value)[source]¶

Cache a computed value for obj

Parameters:: obj – Model object

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

Generic pagination class.

You don’t have to override that class to do basic paginator customization, instead, you can also implement pagination specifics into:

the Client class with the pagination_initialize() and pagination_parameters() methods
or also, per model, in the Model class with the pagination_initialize() and pagination_parameters() methods

Refer to pagination_parameters() and pagination_initialize() for details.

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.

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 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(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, paginator, page_number):
    return dict(page=page_number)

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 cli2.client.RefusedResponseError(client, response, tries, mask, msg=None)[source]¶

class cli2.client.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 cli2.client.ResponseError(client, response, tries, mask, msg=None)[source]¶

enhance(msg)[source]¶

Enhance an httpx.HTTPStatusError

Adds beatiful request/response data to the exception.

Parameters:: exc – httpx.HTTPStatusError

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

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

Example CLI¶

cli2-example-client¶

cli2-example-client ¶: Client for restful-api.dev

cli2-example-client fail¶

cli2-example-client fail ¶

Function: cli2.example_client.APIClient.fail()

Send bogus Form data

cli2-example-client get¶

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

Function: cli2.client.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`

cli2-example-client request¶

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

Function: cli2.client.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`

cli2-example-client object¶

cli2-example-client object ¶

restful-api.dev objects

Example:

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

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

cli2-example-client object find¶

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

Function: cli2.client.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

cli2-example-client object get¶

cli2-example-client object get [KWARGS=VALUE]... ¶

Function: cli2.client.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`

cli2-example-client object delete¶

cli2-example-client object delete ID ¶

Function: cli2.client.Model.delete()

Delete model.

DELETE request on url

Argument	Help
`ID`	ID Required: True

cli2-example-client object create¶

cli2-example-client object create [KWARGS=VALUE]... ¶

Function: cli2.client.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`

cli2-example-client object fail¶

cli2-example-client object fail ¶

Function: cli2.example_client.Object.fail()

Send bogus JSON

cli2-example-client object rename¶

cli2-example-client object rename ID NEW_NAME ¶

Function: cli2.example_client.Object.rename()

Send bogus JSON with an instance

Argument

Help

ID

Required: True

NEW_NAME

Required: True

Tutorial for cli2.Client: HTTP Client framework¶

Example¶

Source code¶

Architecture¶

Tutorial¶

Creating a Client¶

Creating a Model¶

Model.client¶

Model.paginate¶

Fields¶

Nested fields¶

Custom types¶

Expressions¶

Parameterable¶

Related¶

Many Related¶

API¶

Example CLI¶

cli2-example-client¶

cli2-example-client fail¶

cli2-example-client get¶

cli2-example-client request¶

cli2-example-client object¶

cli2-example-client object find¶

cli2-example-client object get¶

cli2-example-client object delete¶

cli2-example-client object create¶

cli2-example-client object fail¶

cli2-example-client object rename¶