Welcome to Bantam’s Documentation!¶

Indices and Tables¶

Introduction¶

Welcome to Bantam, a framework for running a web server, built on top of aiohttp, poviding a convience-layer of abstraction. The framework allows users to define a Python web API through static methods of classes, and allows auto-generation of corresponding javascript APIs to match. The developer need not be concerned about the details of how to map routes nor to understand the details of HTTP transactions. Ths developer need only focus on development of a web API as a Python interface.

Getting Started¶

Let’s look at setting up a simple WebApplication on your localhost:

>>> import asyncio
... from bantam.http import web_api, RestMethod, WebApplication
...
... class Greetings:
...
...     @classmethod
...     @web_api(content_type='text/html', method=RestMethod.GET)
...     async def welcome(cls, name: str) -> str:
...         """
...         Welcome someone
...
...         :@param name: name of person to greet
...         :return: a salutation of welcome
...         """
...         return f"<html><body><p>Welcome, {name}!</p></body></html>"
...
...     @classmethod
...     @web_api(content_type='text/html', method=RestMethod.GET)
...     async def goodbye(cls, type_of_day: str) -> str:
...         """
...         Tell someone goodbye by telling them to have a day (of the given type)
...
...         :@param type_of_day:  an adjective describe what type of day to have
...         :return: a saltation of welcome
...         """
...         return f"<html><body><p>Have a {type_of_day} day!</p></body></html>"
...
... if __name__ == '__main__':
...     app = WebApplication()
...     asyncio.run(app.start()) # default to localhost HTTP on port 8080

Saving this to a file, ‘salutations.py’, you can run this start your server:

% python3 salutations.py

Then open a browser to the following URL’s:

to display various salutiations.

To explain this code, the @web_api decorator declares a method that is mapped to a route. The route is determined by the class name, in this case Greetings, and the method name. Thus, the “welcome” method above, as a member of Greetings class, is mapped to the route ‘/Greetings/welcome”. There are some rules about methods declared as @web_api:

They can be @classmethods or even instance methods (explained below), but @classmethod or instance methods are prefered
They must provide all type hints for parameters and return value. The types must be of only specific kinds as explained below
Be judicious on GET vs POST, particuraly being mindfl of the amount of data to be passed in parameters as wll as returned. Large data transfers should use POST

The query parameters provided in the full URL are mapped to the parameters in the Python method. For example, the query parameter name=Box in the first URL above maps to the name parameter of the Greetings.welcome method, with ‘Bob’ as the value. The query parameters are translated to the value and type expected by the Python API. If the value is not convertible to the proper type, an error code along with reason are returned. There are a few other options for parameters and return type that will be discussed later on streaming.

The methods can also be declared as POST operations. In this case, parameter values would be sent as part of the payload of the request (not query parameters) as a simple JSON dictionary.

Caution

Although the code prevents name collisions, the underlying (automated) routes do not, and a route must be unique. Thus, each pair of class/method declared as a @web_api must be unique, even across differing modules.

Caution

Batnam invokes all requests in a single thread within the server (and undoes any per-thread model of the underlying http/web package used to conduct the HTTP transactions). The user needs to be keenly aware of the rules of asyncio. Specifically: (1) do not invoke blocking calls (await calls that yield processing back but take a long time to complete are OK, of course) and (2) understand that if stateful, the state of the server objects can change in the middle of execution if an await is invoked and processing is yielded to another async request from another client call.

The specific WebApplication API is simple, and only the start method is of importance:

class bantam.http.WebApplication(*, static_path: Path | str | None = None, js_bundle_name: str | None = None, handler_args: Mapping[str, Any] | None = None, client_max_size: int = 1048576, using_async: bool = True, debug: Any = Ellipsis)¶

Main class for running a WebApplication server. See the documentation for aiohttp for information on the various parameters. Additional parameters for this class are:

Parameters:

static_path – root path to where static files will be kept, mapped to a route “/static”, if provided
js_bundle_name – the name of the javascript file, without extension, where the client-side API will be generated, if provided

exception DuplicateRoute¶: Raised if an attempt is made to register a web_api function under an existing route

async start(modules: List[str], host: str | None = None, port: int | None = None, path: str | None = None, initializer: Callable[[], None] | None = None, shutdown_timeout: float = 60.0, ssl_context: SSLContext | None = None, backlog: int = 128, handle_signals: bool = True, reuse_address: bool | None = None, reuse_port: bool | None = None) → None¶

start the app

Parameters:

modules – list of name of modules to import that contain the @web_api definitions to use, must not be empty and will be imported to load the @web_api definitions
host – optional host of app, defaults to ‘127.0.0.1’
port – optional port to listen on (TCP)
path – path, if using UNIX domain sockets to listen on (cannot specify both TCP and domain parameters)
initializer – optional function (no params, no return) to call on first bring-up, inside the thread associated with the app’s asyncio loop
shutdown_timeout – force shutdown if a shutdown call fails to take hold after this many seconds
ssl_context – for HTTPS server; if not provided, will default to HTTP connection
backlog – number of unaccepted connections before system will refuse new connections
handle_signals – gracefully handle TERM signal or not
reuse_address – tells the kernel to reuse a local socket in TIME_WAIT state, without waiting for its natural timeout to expire. If not specified will automatically be set to True on UNIX.
reuse_port – tells the kernel to allow this endpoint to be bound to the same port as other existing endpoints are bound to, so long as they all set this flag when being created. This option is not supported on Windows.

Of particular interest is the modules parameter of the start method. This determines which modules and therefore which classes and their the @web_api definitions are loaded.

bantam.decorators.web_api(content_type: str, method: RestMethod = RestMethod.GET, is_constructor: bool = False, expire_obj: bool = False, on_disconnect: Callable[[], None | Awaitable[None]] | None = None, timeout: ClientTimeout | None = None, uuid_param: str | None = None, preprocess: Callable[[Request], None | Dict[str, Any]] | None = None, postprocess: Callable[[Response | StreamResponse], Response | StreamResponse] | None = None) → Callable[[Callable[[...], Awaitable[Any] | AsyncIterator[Any]]], Callable[[...], Awaitable[Any] | AsyncIterator[Any]]]¶

Decorator for class async method to register it as an API with the WebApplication class Decorated functions should be static class methods with parameters that are convertible from a string (things like float, int, str, bool). Type hints must be provided and will be used to dynamically convert query parameeter strings to the right type.

>>> class MyResource:
...
...   @classmethod
...   @web_api(content_type="text/html")
...   def say_hello(cls, name: str):
...      return f"Hi there, {name}!"

Only GET calls with explicit parameters in the URL are support for now. The above registers a route of the form:

http://<host>:port>/MyRestAPI?name=Jill

Parameters:

content_type – content type to disply (e.g., “text/html”)
method – one of MethodEnum rest api methods (GET or POST)
is_constructor – set to True if API is static method return a class instnace, False oherwise (default)
expire_obj – for instance methods only, epxire the object upon successful completion of that call
on_disconnect – callback if client disconnects unexpectedly
timeout – optional timeout value for response to request to timeout
uuid_param – optional name of parameter to use as unique id for ‘self’
preprocess – optional preprocess function to invoke on request
postprocess – optional postprocess function to run after servicing request

Returns:

callable decorator

Allowed Type Annotations¶

Being http-based, the magic behind the scenes in abstracting the HTTP protocol away from the client occurs through json serialization and deserialization. This means that only certain type hints are allowed in the signature of a method decorated with @web_api:

The basic builtins of int, float, str, bool
For arguments, the builtins of list, set, tuple and dict; the return type must be explicit (e.g. Dict[Union[int, str]]),as the system must know the type when converting fom a generic json string
Anything hat is decorated with @dataclass and whose element types meet these criteria (recursively)
Any typing Dict, List, Set, Tuple, provided the (key and) element type recursively meet these criteria
Use of Union and Optional are allowed

All arguments and the return type must be explicitly specified in the signature of @web_api-decorated methods.

Client-Side Code Generation¶

As an even greater convenience, Bantam provides means of auto-generating both javascript and client code to interact with a bantam-enabled application. Javascript in particular can be generate on-the-fly and served upon startup.

Javascript¶

the code can be generated on the fly, when the WebApplication is created. By providing two parameters to the init call to your WebApplication:

static_path: the root path to serve static files
js_bundle_name: the name (without extension) of the javascript bundle file

the javascript code will be (re)generated just before startup of the server. The javascript file that provides the web API on the client side will then be available under the static route /static/js/<js-bundle-name>.js.

In the example above the instantiations of the app would be:

app = WebApplication(static_path=Path("./static"), js_bundle_name='salutations')

The client interface is then available through:

http://localhost:8080/static/js/salutations.js

Through a simple normal declaration of an API in the Python code, and auto-generation of client code, the developer is free to ignore the details of the inner works of routes and HTTP transactions.

Furthermore, the web api that is registered is also directly callable as a Python function. This engenders the concept of “library-as-a-service”, where the code can act directly as a library or as a distributed Rest-ful service transparently to the caller.

To provide more detail on what this generation look like:

Bantam provides the ability to auto-generate client-side javascript code to abstract the details of makting HTTP requests to the server. This abstraction implies that the developer never need to know about routes, formulating URLs, how to make a POST request, how to stream data over HTTP, etc!

To generate client side code, create a Python source file – let’s name it generator.py – and import all of the classes containing @web_api’s to be generated. Then add the main entry poitn to generate the javascript code, like so:

>>> from bantam.js import JavascriptGenerator
... from salutations import Greetings
...
... if __name__ == '__main__':
...     with open('salutations.js', 'bw') as output:
...        JavascriptGenerator.generate(out=output, skip_html=True)

Run the script as so:

% python generate.py

With the above Greetings example, the generated javasctipt code will mimic the Python:

javascript code auto-generated from Python server’s web API¶

class bantam {};
bantam.salutations = class {};
bantam.salutations.Greetings = class {
      constructor(site){this.site = site;}

      /*
      Welcome someone
      The response will be provided as the (first) parameter passed into onsuccess

      @param {function(string) => null} onsuccess callback inoked, passing in response from server on success
      @param {function(int, str) => null}  onerror  callback upon error, passing in response code and status text
      @param {{string}} name name of person to greet
      */
      welcome(onsuccess, onerror, name) {
         ...
      }

       /*
       Tell someone goodbye by telling them to have a day (of the given type)
       The response will be provided as the (first) parameter passed into onsuccess

       @param {function(string) => null} onsuccess callback inoked, passing in response from server on success
       @param {function(int, str) => null}  onerror  callback upon error, passing in response code and status text
       @param {{string}} type_of_day adjective describing type of day to have
       */
      goodbye(onsuccess, onerror, type_of_day) {
        ...
      }
};

The code maintains the same hierarchy in namespaces as modules in Python, albeit under a global bantam namespace. This prevents potential namespace collisions. The signatures of the API mimic those defined in the Pyhon codebase, with the noted exception of the onsuccess and onerror callbacks as the first two parameters. This is typical of how plain javascript handles asynchronous transactions: rather than returning a value or raising an exception, these callbacks are invoked instead, upon response from the server.

Python¶

Bantam provides and abstraction to easily declare Python clients to interact with a Bantam web application.

To access client code, a server implementation must inherit from an abstract interface common to both client and server. One defines an interface class (usually in its own file) such as:

>>>  from bantam.client import WebInterface
...  from bantam.api import RestMethod
...  from bantam.decorators web_api
...  from typing import AsyncIterator
...  from abc import abstractmethod
...
...  class MyServerApiInterface(WebInterface):
...     @classmethod
...     @web_api(methos=RestMethod.GET, content_type='application/json')
...     @abstractmethod
...     async def constructor(cls) -> "MyServerApiInterface":
...        '''
...        Abstract constructor to create an instance of the class
...        '''
...
...      @classmethod
...      @web_api(method=RestMethod.GET, content_type='text/plain')
...      @abstractmethod
...      async def class_method_api(cls, parm1: str) -> Dict[str, int]:
...          '''
...           Abstract class method to be implemented WITH SAME @web_api DECORATION
...          '''
...
...      @web_api(method=RestMethod.POST, content_type='application/json')
...      @abstractmethod
...      async def instance_method_api(self) -> AsyncIterator[str]:
...          '''
...          Abstract instance method that is an Async Iterator over string values. MUST
...          HAVE SAME @web_api DECORATION AS SERVER IMPL DECLARATION
...          '''
...          # never called since abstract, but needed to tell Python that implementation
...          # will be an async generator/iterator:
...          yield None
...

One then defines the concrete class (which for best practice, has same name, sans ‘Interface’). The @web_api decorators need not be specified as bantam will ensure they are inherited. (But if you do specify them explicitly, you must ensure they are maintained to be the same):

>>>  from bantam.client import WebInterface
...  from bantam.api import RestMethod
...  from bantam.decorators web_api
...  from typing import AsyncIterator
...
...  class MyServerApi(MyServerApiInterface):
...
...     def __init__(self):
...         self._message_bits = ['I', 'am', 'the', 'very', 'model']
...
...     @classmethod
...     async def constructor(cls) -> "MyServerApiInterface":
...        '''
...        Concreate constructor to create an instance of the class WITH SAME @web_api DECORATOR AND
...        SIGNATURE
...        '''
...        return MyServerApi()
...
...      @classmethod
...      async def class_method_api(cls, parm1: str) -> Dict[str, int]:
...          '''
...          Concreate class method to be implemented WITH SAME @web_api DECORATION AND,
...          OF COURSE, SIGNATURE
...          '''
...          return {'param1': int(parm1)}
...
...      async def instance_method_api(self) -> AsyncIterator[str]:
...          '''
...          Concrete instance method that is an Async Iterator over string values.
...          MUST HAVE SAME @web_api DECORATION AND SIGNATURE
...          '''
...          # never called since abstract, but needed to tell Python that implementation
...          # will be an async generator/iterator:
...          for text in self._message_bits:
...              yield text
...

One can then declare a Client that acts as a proxy to make calls to the server, thereby keeping the http protocol details hidden (abstracted away frm the user):

>>>  def async_call():
...      client_end_point_mapping = MyServerApiInterface.ClientEndpointMapping()
...      client = await client_end_point_mapping['https://blahblahblah']
...      instance = client.constructor()
...      async for text in instance.instance_method_api():
...          print(text)
...

If you do not fallow the recommended practice of the interface have the same name as the concreate class, only with an “Interface” suffix, you will have to specify impl_name=<name-of-concrete-class> as a parameter to Client class method above. The end_point parameter specifies the base url to the server that serves up MyServceApi class.

Advanced Discussion on Parameters¶

Optional Parameters¶

If default parameter values are provided, they are optionally provided on the javascript client-side as well. Additionally, to refine the second rule of declaring a web API above, the type for a parameter can also be declared as an Optional to any the types declared there. Provided a default value, include None, is provided.

Streamed Responses¶

Responses can be streamed back to the client. To do so, the underlying Python server call must be written as an AsyncGenerator, returning a type AsyncGenerator[None, <int, float, bool, str, or bytes>]. (Thus, ammending the rule of allowed return values above). Consider this example of a PoetryReader class:

Example of server code to stream a poem back to the client¶

 import asyncio
 import os

 from bantam.decorators import RestMethod
 from bantam.web import WebApplication, web_api
 from typing import AsyncGenerator

 class PoetryReader:
     POEM = """
     Mary had a little lamb...
     Its fleece was white as snow...
     Everywhere that Mary went...
     Her lamb was sure to go...
     """

     @classmethod
     @web_api(content_type='text/streamed', method=RestMethod.POST)
     async def read_poem(cls) -> AsyncGenerator[None, str]:
         for line in PoetryReader.POEM.strip().splitlines():
             yield line
             await asyncio.sleep(2)  # for dramatic effect
         yield "THE END"
         await asyncio.sleep(2)

 if __name__ == '__main__':
     app = WebApplication(static_path=os.path.dirname(__file__), js_bundle_name='poetry_reader')
     asyncio.get_event_loop().run_until_complete(app.start())  # default to localhost HTTP on port 8080

Caution

The method must be declared as POST to work consistently across different machines/web browsers

The javascript generated for such an API is the same, with the exception that the onsuccess callback will be named onreceive and will be called with two parameters. The first is the “chunk” that is read and the second is a boolean indicating whether it is the final one or not. The “chunks” are determine by the type yielded by the generator:

for numeric or bool types: each number yielded will invoke a single callback to onreceive. But be aware that multiple
values can be sent at a time and that the generated javascript code will handle the logic to split them up. This is particularly true when there is no or short sleep periods between yields.
for str type: onreceive will be called for each line received, guaranteeing whole lines are provided
for bytes type: cnreceive will be called for each chunk of bytes received, regardless of size

Here is a bit of HTML to implement the client-side:

HTML containing client-side code for handling streaming responses¶

 <html>
 <head>
     <script src="/static/js/poetry_reader.js" ></script>
 </head>
 <body>
 <div id='poem'></div>
 <script>
     document.addEventListener('DOMContentLoaded', function(event) {

         let reader = new bantam.__main__.PoetryReader("http://localhost:8080");
         let html = ""
         let onreceive = function(line, is_done){
              html += "<p>" + line + "</p><br/>";
              document.getElementById('poem').innerHTML = html;
         }
         reader.read_poem(onreceive, function(code, reason){
             alert("UNEXPECTED ERROR OCCURRED: " + code + " : " + reason);
         });
     });
 </script>
 </body>
 </html>

Loading the page http://localhost:8080/static/index.html will execute the code.

Streamed Requests¶

You can also make requests that stream (upload) data to the server. Here again, we add two more (final) allowable types for parameters that specify a parameter provides an (async) iterator for lopping over and sending data chunk by chunk:

bantam.web.AsyncLineGIterator: specifies that the client will send str data to the server line by line (without any return character present in the line)
bantam.web.AsyncChunkIterator: specifies that the iterator will send bytes of data to the server

Caution

At most one parameter may be specified as an iterator type

Here is what a web API method would look like that captures uploaded byts of data to a file and reports progress back to the client:

Example code for Python web API method that streams data¶

 class Uploader:
     @classmethod
     @web_api(content_type='text/html')
     async def receive_streamed_data(cls, size: int, byte_data: AsyncChunkGenerator) -> AsyncGenerator[None, float]
         bytes_received: int = 0
         packet_size = 100 # bytes
         with open('some_file_path', 'bw') as f:
             async for data in byte_data(packet_size):
                 f.write(byte_data)
                 bytes_received += len(byte_data)
                 progress = bytes_received*100/size
                 yield progress

As a side note, when a streaming parameter is specified, all others will be sent as query parameters even when the request is POST. Note that for chunk iterators, the parameter byte_data will be a callable that takes a single parameter, which is the max size of packet to be read each chunk. The line iterator will be a simple iterator, not a callable. That is for a line itertor the Python code would look like:

...
class Uploader:
    @classmethod
    @web_api(content_type='text/html')
    async def receive_streamed_data(cls, size: int, line_by_line: AsyncLineGenerator) -> AsyncGenerator[None, float]
        ...
        with open('some_file_path', 'bw') as f:
            async for data in line_by_line:  #<====  NOT a function call as in AsyncChunkIterator
               ...

On the client side, the signature of the API will look similar, with the onsuccess (onreceive if the response is also streaming) and onerror callbacks act the same. The differences are that:

the parameter that is the AsyncLine[Chunk]Iterator will not be present in the signature
insted, the javascript API call will return a function to be called by the client to send each chunk of data (ad the first and only parameter of the function)

An example of uploading a file using the above api (assuming the Uploader class belongs to the module uploads):

...
let onreceive = function(progress, is_done){
   ...
}
let onerror = function(code, reason){
   ...
}
let uploader = bantam.uploads.Uploader(site_url)
let size_of_upload = 1024*1024;  // bytes
let send_byte_data = uploader.receive_streamed_data(onreceive, onerror, size_of_upload)
const reader = new FileReader();
reader.addEventListener('load', (event) => {
    send_byte_data(event.target.result);
});
reader.readAsText(file); // assumng file is a file object created prior to this code

Pre/Post-Processing Request/Responses¶

You can pre-process requests before a call is made to the underlying Python web API. Likewise, you can modify responses returned from a Python web API call before it is sent to the client. There are two means to do so. One is at the application level, by calling set_preprocessor and set_postprocessor accordingly:

def preprocessor(request: Request) -> Union[None, Dict[str, Any]]:
   return {'additional_arg': "value"}  # adds an additional argument to the web API call; can also return None

def postprocessor(response: Response) -> None:
   response.body += b"\n Additional line of output"

...

app = WebApplication(...)
app.set_preprocessing(preprocessor)
app.set_postprocessing(postprocessor)

Alternatively, you can do this for each individual web API method through optional preprocess and postprocess arguments. Using the above define prerprocessor and postprocessor functions, this would look like:

...
@web_api(content_type='text/plain', preprocess=preprocessor, postprocess=postprocessor)

Getting Request Context¶

In the implementation of a web api, you can get the context of the rquest (i.e., the request headers). Such information may be useful if chaining one Rest call to other backend Rest calls in a service hiearchy, and passing any contextual information around the originating request down through such a chain of calls. Context is available through the WebApplication.get_context() call which returns a dictionary of the reques header keys and values.

Be aware that accessing context may limit the use of your code as a “library-as-a-service”. If a direct call is made to a web api without any request, the get_context() call will return None, a case you should consider handling.

Welcome to Bantam’s Documentation!¶

Indices and Tables¶

Introduction¶

Getting Started¶

Allowed Type Annotations¶

Client-Side Code Generation¶

Javascript¶

Python¶

Advanced Discussion on Parameters¶

Optional Parameters¶

Streamed Responses¶

Streamed Requests¶

Pre/Post-Processing Request/Responses¶

Getting Request Context¶

Table of Contents

This Page