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,

• export HTTP_DEBUG=1 for low-level HTTP Debugging output

• a ORM for REST resources

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:

• 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.

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):
    def __init__(self, *args, **kwargs):
        kwargs.setdefault('base_url', 'https://api.restful-api.dev/')
        super().__init__(*args, **kwargs)


class Object(APIClient.Model):
    url_list = '/objects'
    url_detail = '/objects/{self.id}'

    id = cli2.Field()

    @classmethod
    @APIClient.cli.cmd
    async def fail(cls):
        await cls.client.post('/foo', json=[1])


cli = APIClient.cli

Result CLI

cli2-example-client

cli2-example-client

HTTPx Client

paginator

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

Sub-Command	Help
help	Get help for a command or group
get	GET Request
fail
object	You should never call this class directly, instead, get it from the Client object after decorating your model classes with a client as such:

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 fail

cli2-example-client fail

Function: cli2.example_client.Object.fail()

cli2-example-client object

cli2-example-client object

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.

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.

url

Object URL based on url_detail and.

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. Example:
delete	Delete model DELETE request on url

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

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

API

HTTP Client boilerplate code to conquer the world.

class cli2.client.Client(*args, **kwargs)

HTTPx Client

paginator

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

class Model(data=None, **values)

property client

Return last client object used, unless it raised RemoteProtocolError.

async delete(url, *args, **kwargs)

DELETE Request

async get(url, *args, **kwargs)

GET Request

async head(url, *args, **kwargs)

HEAD Request

paginate(url, params=None, model=None)

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)

Implement Model-specific pagination initialization here.

Refer to Paginator.pagination_initialize() for details.

pagination_parameters(paginator, page_number)

Implement Model-specific pagination parameters here.

Refer to Paginator.pagination_parameters() for details.

paginator

alias of Paginator

async post(url, *args, **kwargs)

POST Request

async put(url, *args, **kwargs)

PUT Request

async request(method, url, **kwargs)

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.

async request_safe(*args, **kwargs)

Request method that retries with a new client if RemoteProtocolError.

async token_get()

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: it’s up to you to call something like:

async def token_get(self):
    response = await self.post('/login', dict(...))
    token = response.json()['token']
    self.client.headers['token'] = f'Bearer {token}'
    return token

class cli2.client.DateTimeField(*args, fmt=None, fmts=None, **kwargs)

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)

Convert the internal string into an external datetime.

internalize(obj, value)

Convert a datetime into an internal string.

class cli2.client.Field(data_accessor=None, parameter=None)

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)

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)

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)

Get the “raw” value from the object, which is a Python representation of the JSON internal value, using data_accessor.

internal_set(obj, value)

Using data_accessor, set the value in Model.data

internalize(obj, value)

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)

Tell the model that the data must be cleaned.

class cli2.client.JSONStringField(*args, options=None, **kwargs)

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)

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)

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)

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.

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.

url

Object URL based on url_detail and.

property data

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

async delete()

Delete model

DELETE request on url

classmethod find(*expressions, **params)

Find objects filtered by GET params

Parameters:

• params – GET URL parameters

• expressions – Expression list

async classmethod get(**kwargs)

Instanciate a model with kwargs and run hydrate().

async hydrate()

Refresh data with GET requset on url_detail

model_command

alias of ModelCommand

model_group

alias of ModelGroup

classmethod paginate(url, *expressions, **params)

Return a Paginator based on url_list :param expressions: Expression list

classmethod pagination_initialize(paginator, data)

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)

Implement Model-specific pagination parameters here.

Otherwise, Client.pagination_parameters() will take place.

Refer to Paginator.pagination_parameters() for details.

paginator

alias of Paginator

class cli2.client.ModelCommand(target, *args, **kwargs)

Command class for Model class.

setargs()

Reset arguments.

class cli2.client.ModelGroup(cls)

class cli2.client.MutableField(data_accessor=None, parameter=None)

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)

Return cached value for obj

Parameters:

obj – Model object

cache_set(obj, value)

Cache a computed value for obj

Parameters:

obj – Model object

class cli2.client.Paginator(client, url, params=None, model=None, expressions=None)

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 initialize(response=None)

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

Parameters:

response – First response object

async page_items(page_number)

Return the items of a given page number.

Parameters:

page_number – Page number to get the items from

async page_response(page_number)

Return the response for a page.

Parameters:

page_number – Page number to get the items from

pagination_initialize(data)

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)

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)

Parse a response and return a list of model items.

Parameters:

response – Response to parse

reverse()

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.

class cli2.client.Related(model, *args, **kwargs)

Related model field.

model

STRING name of the related model class.

externalize(obj, value)

Instanciate the related model class with the value.

internalize(obj, data)

Return the related object’s data.