rororo API

OpenAPI

rororo.openapi

Cornerstone of rororo library, which brings OpenAPI 3 schema support for aiohttp.web applications.

rororo.openapi.setup_openapi(app, schema_path, *operations, server_url=None, is_validate_response=True, has_openapi_schema_handler=True, use_error_middleware=True, error_middleware_kwargs=None, use_cors_middleware=True, cors_middleware_kwargs=None)[source]

Setup OpenAPI schema to use with aiohttp.web application.

Unlike aiohttp-apispec and other tools, which provides OpenAPI/Swagger support for aiohttp.web applications, rororo changes the way of using OpenAPI schema with aiohttp.web apps.

rororo relies on concrete OpenAPI schema file, path to which need to be registered on application startup (mostly inside of create_app factory or right after aiohttp.web.Application instantiation).

And as valid OpenAPI schema ensure unique operationId used accross the schema rororo uses them as a key while telling aiohttp.web to use given view handler for serving required operation.

With that in mind registering (setting up) OpenAPI schema requires:

  1. aiohttp.web.Application instance

  2. Path to file (json or yaml) with OpenAPI schema

  3. OpenAPI operation handlers mapping (rororo’s equialent of aiohttp.web.RouteTableDef)

In common cases setup looks like,

from pathlib import Path
from typing import List

from aiohttp import web

from .views import operations


def create_app(argv: List[str] = None) -> web.Application:
    return setup_openapi(
        web.Application(),
        Path(__file__).parent / "openapi.yaml",
        operations,
    )

If your OpenAPI schema contains multiple servers schemas, like,

servers:
- url: "/api/"
  description: "Test environment"
- url: "http://localhost:8080/api/"
  description: "Dev environment"
- url: "http://prod.url/api/"
  description: "Prod environment"

you have 2 options of telling rororo to use specific server URL.

First, is passing server_url, while setting up OpenAPI, for example,

setup_openapi(
    web.Application(),
    Path(__file__).parent / "openapi.yaml",
    operations,
    server_url=URL("http://prod.url/api/"),
)

Second, is more complicated as you need to wrap aiohttp.web application into rororo.settings.setup_settings() and mark each server with x-rororo-level special key in server schema definition as,

servers:
- url: "/api/"
  x-rororo-level: "test"
- url: "http://localhost:8080/api/"
  x-rororo-level: "dev"
- url: "http://prod.url/api/"
  x-rororo-level: "prod"

After, rororo will try to equal current app settings level with the schema and if URL matched, will use given server URL for finding out route prefix.

By default, rororo will validate operation responses against OpenAPI schema. To disable this feature, pass is_validate_response falsy flag.

By default, rororo will share the OpenAPI schema which is registered for your aiohttp.web application. In case if you don’t want to share this schema, pass has_openapi_schema_handler=False on setting up OpenAPI.

By default, rororo will enable aiohttp_middlewares.cors.cors_middleware() without any settings and aiohttp_middlewares.error.error_middleware() with custom error handler to ensure that security / validation errors does not provide any mess to command line. Pass use_cors_middleware / use_error_middleware to change or entirely disable this default behaviour.

For passing custom options to CORS middleware, use cors_middleware_kwargs mapping. If kwarg does not support by CORS middleware - rororo will raise a ConfigurationError. All list of options available at documentation for aiohttp_middlewares.cors.cors_middleware().

Return type

Application

class rororo.openapi.OperationTableDef[source]

Map OpenAPI 3 operations to aiohttp.web view handlers.

In short it is rororo’s equialent to aiohttp.web.RouteTableDef. Under the hood, on rororo.openapi.setup_openapi() it still will use RouteTableDef for registering view handlers to aiohttp.web.Application.

But unlike RouteTableDef it does not register any HTTP method handlers (as via @routes.get decorator) in favor of just registering the operations.

There are two ways for registering view hanlder,

  1. With bare @operations.register call when OpenAPI operationId equals to view handler name.

  2. Or by passing operation_id to the decorator as first arg, when operationId does not match view handler name, or if you don’t like the fact of guessing operation ID from view handler name.

Both of ways described below,

from rororo import OperationTableDef

operations = OperationTableDef()

# Expect OpenAPI 3 schema to contain operationId: hello_world
@operations.register
async def hello_world(request: web.Request) -> web.Response:
    ...

# Explicitly use operationId: helloWorld
@operations.register("helloWorld")
async def hello_world(request: web.Request) -> web.Response:
    ...

If supplied operation_id does not exist in OpenAPI 3 schema, rororo.openapi.setup_openapi() call raises an OperationError.

rororo.openapi.openapi_context(request)[source]

Context manager to access valid OpenAPI data for given request.

If request validation done well and request to OpenAPI operation view handler is valid one, view handler may need to use request data for its needs. To achieve it use given context manager as,

from rororo import openapi_context, OperationTableDef

operations = OperationTableDef()

@operations.register
async def hello_world(request: web.Request) -> web.Response:
    with openapi_context(request) as context:
        ...

OpenAPIContext class will contain next attributes:

  • request

  • app

  • operation

  • parameters

  • data

If using context managers inside of view handlers considered as unwanted, there is an other option in rororo.openapi.get_openapi_context() function.

Return type

Iterator[OpenAPIContext]

rororo.openapi.get_openapi_context(request)[source]

Shortcut to retrieve OpenAPI schema from aiohttp.web request.

OpenAPIContext attached to aiohttp.web.Request instance only if current request contains valid data.

ContextError raises if, for some reason, the function called outside of valid OpenAPI request context.

Return type

OpenAPIContext

rororo.openapi.get_openapi_schema(mixed)[source]

Shortcut to retrieve OpenAPI schema from aiohttp.web application.

ConfigruationError raises if aiohttp.web.Application does not contain registered OpenAPI schema.

Return type

Dict[str, Any]

rororo.openapi.get_openapi_spec(mixed)[source]

Shortcut to retrieve OpenAPI spec from aiohttp.web application.

ConfigruationError raises if aiohttp.web.Application does not contain registered OpenAPI spec.

Return type

Spec

rororo.openapi.get_validated_data(request)[source]

Shortcut to get validated data (request body) for given request.

In case when current request has no valid OpenAPI context attached - ContextError will be raised.

Return type

Any

rororo.openapi.get_validated_parameters(request)[source]

Shortcut to get validated parameters for given request.

In case when current request has no valid OpenAPI context attached - ContextError will be raised.

Return type

OpenAPIParameters

Settings

rororo.settings

Useful functions to work with application settings such as,

  • Locale

  • Logging

  • Time zone

As well as provide attrib factory helper to read settings from environment to use within Settings data structures.

class rororo.settings.BaseSettings(host=NOTHING, port=NOTHING, debug=NOTHING, level=NOTHING, time_zone=NOTHING, first_weekday=NOTHING, locale=NOTHING, sentry_dsn=NOTHING, sentry_release=NOTHING)[source]

Base Settings data structure for configuring aiohttp.web apps.

Provides common attribs, which covers most of settings requires for run and configure aiohttp.web app. In same time it is designed to be inherited and completed with missed values in application as,

import attr
from rororo.settings import BaseSettings, env_factory


@attr.dataclass(frozen=True, slots=True)
class Settings(BaseSettings):
    other_name: str = env_factory("OTHER_NAME", "other-value")
rororo.settings.env_factory(name, default=None)[source]

Helper to read attribute value from environment.

It is designed to use, when settings is implemented as @attr.dataclass data structure and there is a need to read value from environment via os.getenv() function.

This factory function helps to:

  1. Eliminate necessity of using lambda functions when mixing os.getenv with attr, like: attr.Factory(lambda: os.getenv("USER"))

  2. Exclude # type: ignore from settings data structures

  3. Convert str values from environment to required type

Example belows demonstrates env_factory usage variants,

import attr


@attr.dataclass
class Settings:
    # As ``os.getenv(key)`` returns ``Optional[str]`` you need
    # to provide default value for each non-optional env attribs
    user: str = env_factory("USER", "default-user")

    # Convert string value from environment to ``Path``
    user_path: Path = env_factory("USER_PATH", Path("~"))

    # Or to integer
    user_level: int = env_factory("USER_LEVEL", 1)

    # When default value is omit env attrib is **optional**
    sentry_dsn: Optional[str] = env_factory("SENTRY_DSN")
Return type

Union[str, None, ~T]

rororo.settings.setup_settings(app, settings, *, loggers=None, remove_root_handlers=False)[source]

Shortcut for applying settings for given aiohttp.web app.

After applying, put settings to aiohttp.web.Application dict as "settings" key.

Return type

Application

rororo.settings.setup_locale(lc_all, first_weekday=None, *, lc_collate=None, lc_ctype=None, lc_messages=None, lc_monetary=None, lc_numeric=None, lc_time=None)[source]

Shortcut helper to setup locale for backend application.

Parameters
  • lc_all (str) – Locale to use.

  • first_weekday (Optional[int]) – Weekday for start week. 0 for Monday, 6 for Sunday. By default: None

  • lc_collate (Optional[str]) – Collate locale to use. By default: <lc_all>

  • lc_ctype (Optional[str]) – Ctype locale to use. By default: <lc_all>

  • lc_messages (Optional[str]) – Messages locale to use. By default: <lc_all>

  • lc_monetary (Optional[str]) – Monetary locale to use. By default: <lc_all>

  • lc_numeric (Optional[str]) – Numeric locale to use. By default: <lc_all>

  • lc_time (Optional[str]) – Time locale to use. By default: <lc_all>

Return type

str

rororo.settings.setup_logging(config, *, remove_root_handlers=False)[source]

Wrapper around logging.config.dictConfig() function.

In most cases it is not necessary to use an additional wrapper for setting up logging, but if your aiohttp.web application run as:

python -m aiohttp.web api.app:create_app

aiohttp will setup logging via logging.basicConfig() call and it may result in duplicated logging messages. To avoid duplication, it is needed to remove logging.root handlers.

Parameters

remove_root_handlers (bool) – Remove logging.root handlers if any. By default: False

Return type

None

rororo.settings.setup_timezone(timezone)[source]

Shortcut helper to configure timezone for backend application.

Parameters

timezone (str) – Timezone to use, e.g. “UTC”, “Europe/Kiev”.

Return type

None

rororo.settings.immutable_settings(defaults, **optionals)[source]

Initialize and return immutable Settings dictionary.

Settings dictionary allows you to setup settings values from multiple sources and make sure that values cannot be changed, updated by anyone else after initialization. This helps keep things clear and not worry about hidden settings change somewhere around your web application.

Deprecated since version 2.0: Function deprecated in favor or using attrs or dataclasses for declaring settings classes. Will be removed in 3.0.

Parameters
  • defaults (Union[module, Dict[str, Any]]) – Read settings values from module or dict-like instance.

  • **optionals

    Update base settings with optional values.

    In common additional values shouldn’t be passed, if settings values already populated from local settings or environment. But in case of using application factories this makes sense:

    from . import settings
    
    def create_app(**options):
        app = ...
        app.settings = immutable_settings(settings, **options)
        return app
    

    And yes each additional key overwrite default setting value.

Return type

Mapping[str, Any]

rororo.settings.is_setting_key(key)[source]

Check whether given key is valid setting key or not.

Only public uppercase constants are valid settings keys, all other keys are invalid and shouldn’t present in Settings dict.

Valid settings keys

DEBUG
SECRET_KEY

Invalid settings keys

_PRIVATE_SECRET_KEY
camelCasedSetting
rel
secret_key
Parameters

key (str) – Key to check.

Return type

bool

rororo.settings.inject_settings(mixed, context, fail_silently=False)[source]

Inject settings values to given context.

Parameters
  • mixed (Union[str, module, Dict[str, Any]]) – Settings can be a string (that it will be read from Python path), Python module or dict-like instance.

  • context (MutableMapping[str, Any]) – Context to assign settings key values. It should support dict-like item assingment.

  • fail_silently (bool) – When enabled and reading settings from Python path ignore errors if given Python path couldn’t be loaded.

Return type

None

rororo.settings.iter_settings(mixed)[source]

Iterate over settings values from settings module or dict-like instance.

Parameters

mixed (Union[module, Dict[str, Any]]) – Settings instance to iterate.

Return type

Iterator[Tuple[str, Any]]

rororo.settings.from_env(key, default=None)[source]

Shortcut for safely reading environment variable.

Deprecated since version 2.0: Use os.getenv() instead. Will be removed in 3.0.

Parameters
  • key (str) – Environment var key.

  • default (Optional[~T]) – Return default value if environment var not found by given key. By default: None

Return type

Union[str, ~T, None]

Logger

rororo.logger

Logging utilities to simplify setting up Python logging.

rororo.logger.default_logging_dict(*loggers, **kwargs)[source]

Prepare logging dict for logging.config.dictConfig().

rororo minds to simplify and unify logging configuration for aiohttp.web applications and cause of that the resulted logging config will:

  • Only messages from loggers will be processed

  • Pass all DEBUG & INFO logging messages to stdout

  • Pass all other messages to stderr

  • Any logging message will be formatted as: "%(asctime)s [%(levelname)s:%(name)s] %(message)s"

For example, to enable logging for aiohttp & api loggers,

from logging.config import dictConfig
dictConfig(default_logging_dict("aiohttp", "api"))
Parameters
  • *loggers – Enable logging for each logger in sequence.

  • **kwargs – Setup additional logger params via keyword arguments.

Return type

Dict[str, Any]

rororo.logger.update_sentry_logging(logging_dict, sentry_dsn, *loggers, level=None, **kwargs)[source]

Enable Sentry logging if Sentry DSN passed.

Deprecated since version 2.0: Deprecated in favor of sentry-sdk and will be removed in 3.0.

Note

Sentry logging requires raven library to be installed.

Usage

from logging.config import dictConfig

LOGGING = default_logging_dict()
SENTRY_DSN = '...'

update_sentry_logging(LOGGING, SENTRY_DSN)
dictConfig(LOGGING)

Using AioHttpTransport for SentryHandler

This will allow to use aiohttp.client for pushing data to Sentry in your aiohttp.web app, which means elimination of sync calls to Sentry.

from raven_aiohttp import AioHttpTransport
update_sentry_logging(
    LOGGING,
    SENTRY_DSN,
    transport=AioHttpTransport
)
Parameters
  • logging_dict (Dict[str, Any]) – Logging dict.

  • sentry_dsn (Optional[str]) – Sentry DSN value. If None do not update logging dict at all.

  • *loggers – Use Sentry logging for each logger in the sequence. If the sequence is empty use Sentry logging to each available logger.

  • **kwargs – Additional kwargs to be passed to SentryHandler.

Return type

None

class rororo.logger.IgnoreErrorsFilter[source]

Ignore all warnings and errors from stdout handler.

filter(record)[source]

Allow only debug and info log messages to stdout handler.

Return type

bool

aio-libs Utils

rororo.aio

Various utilities for aiohttp and other aio-libs.

rororo.aio.add_resource_context(router, url_prefix=None, name_prefix=None)[source]

Context manager for adding resources for given router.

Main goal of context manager to easify process of adding resources with routes to the router. This also allow to reduce amount of repeats, when supplying new resources by reusing URL & name prefixes for all routes inside context manager.

Behind the scene, context manager returns a function which calls:

resource = router.add_resource(url, name)
resource.add_route(method, handler)

For example to add index view handler and view handlers to list and create news:

with add_resource_context(app.router, "/api", "api") as add_resource:
    add_resource("/", get=views.index)
    add_resource("/news", get=views.list_news, post=views.create_news)
Parameters
  • router (UrlDispatcher) – Route to add resources to.

  • url_prefix (Optional[str]) – If supplied prepend this prefix to each resource URL.

  • name_prefix (Optional[str]) – If supplied prepend this prefix to each resource name.

Return type

Iterator[AddResourceFunc]

rororo.aio.is_xhr_request(request)[source]

Check whether current request is XHR one or not.

Basically it just checks that request contains X-Requested-With header and that the header equals to XMLHttpRequest.

Parameters

request (Request) – Request instance.

Return type

bool

rororo.aio.parse_aioredis_url(url)[source]

Convert Redis URL string to dict suitable to pass to aioredis.create_redis(...) call.

async def connect_redis(url=None):
    url = url or "redis://localhost:6379/0"
    return await create_redis(**parse_aioredis_url(url))
Parameters

url (str) – URL to access Redis instance, started with redis://.

Return type

Dict[str, Any]

Timedelta Utils

rororo.timedelta

Useful functions to work with timedelta instances.

rororo.timedelta.str_to_timedelta(value, fmt=None)[source]

Convert string value to timedelta instance according to the given format.

If format not set function tries to load timedelta using default TIMEDELTA_FORMAT and then both of magic “full” formats.

You should also specify list of formats and function tries to convert to timedelta using each of formats in list. First matched format would return the converted timedelta instance.

If user specified format, but function cannot convert string to new timedelta instance - ValueError would be raised. But if user did not specify the format, function would be fail silently and return None as result.

Parameters
  • value (str) – String representation of timedelta.

  • fmt (Optional[str]) – Format to use for conversion.

Return type

Optional[timedelta]

rororo.timedelta.timedelta_average(*values)[source]

Compute the arithmetic mean for timedeltas list.

Parameters

*values – Timedelta instances to process.

Return type

timedelta

rororo.timedelta.timedelta_div(first, second)[source]

Implement divison for timedelta instances.

Parameters
  • first (timedelta) – First timedelta instance.

  • second (timedelta) – Second timedelta instance.

Return type

Optional[float]

rororo.timedelta.timedelta_seconds(value)[source]

Return full number of seconds from timedelta.

By default, Python returns only one day seconds, not all timedelta seconds.

Parameters

value (timedelta) – Timedelta instance.

Return type

int

rororo.timedelta.timedelta_to_str(value, fmt=None)[source]

Display the timedelta formatted according to the given string.

You should use global setting TIMEDELTA_FORMAT to specify default format to this function there (like DATE_FORMAT for builtin date template filter).

Default value for TIMEDELTA_FORMAT is 'G:i'.

Format uses the same policy as Django date template filter or PHP date function with several differences.

Available format strings:

Format character

Description

Example output

a

Not implemented.

A

Not implemented.

b

Not implemented.

B

Not implemented.

c

Not implemented.

d

Total days, 2 digits with leading zeros. Do not combine with w format.

'01', '41'

D

Not implemented.

f

Magic “full” format with short labels.

'2w 4d 1:28:07'

F

Magic “full” format with normal labels.

'2 weeks, 4 days, 1:28:07'

g

Day, not total, hours without leading zeros. To use with d, j, or w.

'0' to '23'

G

Total hours without leading zeros. Do not combine with g or h formats.

'1', '433'

h

Day, not total, hours with leading zeros. To use with d or w.

'00' to '23'

H

Total hours with leading zeros. Do not combine with g or h formats.

'01', ``'433'

i

Hour, not total, minutes, 2 digits with leading zeros To use with g, G, h or H formats.

00 to '59'

I

Total minutes, 2 digits or more with leading zeros. Do not combine with i format.

'01', '433'

j

Total days, one or 2 digits without leading zeros. Do not combine with w format.

'1', '41'

J

Not implemented.

l

Days long label. Pluralized and localized.

'day' or 'days'

L

Weeks long label. Pluralized and localized.

'week' or 'weeks'

m

Week days long label. Pluralized and localized.

'day' or 'days'

M

Not implemented.

n

Not implemented.

N

Not implemented.

O

Not implemented.

P

Not implemented.

r

Standart Python timedelta representation with short labels.

'18 d 1:28:07'

R

Standart Python timedelta representation with normal labels.

'18 days, 1:28:07'

s

Minute, not total, seconds, 2 digits with leading zeros. To use with i or I.

'00' to '59'

S

Total seconds. 2 digits or more with leading zeros. Do not combine with s format.

'00', '433'

t

Not implemented.

T

Not implemented.

u

Second, not total, microseconds.

0 to 999999

U

Not implemented.

w

Week, not total, days, one digit without leading zeros. To use with W.

0 to 6

W

Total weeks, one or more digits without leading zeros.

'1', '41'

y

Not implemented.

Y

Not implemented.

z

Not implemented.

Z

Not implemented.

For example,

>>> import datetime
>>> from rororo.timedelta import timedelta_to_str
>>> delta = datetime.timedelta(seconds=99660)
>>> timedelta_to_str(delta)
... '27:41'
>>> timedelta_to_str(delta, 'r')
... '1d 3:41:00'
>>> timedelta_to_str(delta, 'f')
... '1d 3:41'
>>> timedelta_to_str(delta, 'W L, w l, H:i:s')
... '0 weeks, 1 day, 03:41:00'

Couple words about magic “full” formats. These formats show weeks number with week label, days number with day label and seconds only if weeks number, days number or seconds greater that zero.

For example,

>>> import datetime
>>> from rororo.timedelta import timedelta_to_str
>>> delta = datetime.timedelta(hours=12)
>>> timedelta_to_str(delta, 'f')
... '12:00'
>>> timedelta_to_str(delta, 'F')
... '12:00'
>>> delta = datetime.timedelta(hours=12, seconds=30)
>>> timedelta_to_str(delta, 'f')
... '12:00:30'
>>> timedelta_to_str(delta, 'F')
... '12:00:30'
>>> delta = datetime.timedelta(hours=168)
>>> timedelta_to_str(delta, 'f')
... '1w 0:00'
>>> timedelta_to_str(delta, 'F')
... '1 week, 0:00'
Parameters
  • value (timedelta) – Timedelta instance to convert to string.

  • fmt (Optional[str]) – Format to use for conversion.

Return type

str

Other Utilities

rororo.utils

Different utility functions, which are common used in web development, like converting string to int or bool.

rororo.utils.ensure_collection(value)[source]

Ensure that given value is a collection, not a single string.

As passing single string validates Collection[str] type annotation, this function converts given single string to a tuple with one item.

In all other cases return given value.

Return type

Collection[str]

rororo.utils.to_bool(value)[source]

Convert string or other Python object to boolean.

Rationalle

Passing flags is one of the most common cases of using environment vars and as values are strings we need to have an easy way to convert them to boolean Python value.

Without this function int or float string values can be converted as false positives, e.g. bool('0') => True, but using this function ensure that digit flag be properly converted to boolean value.

Parameters

value (Any) – String or other value.

Return type

bool

rororo.utils.to_int(value, default=None)[source]

Convert given value to int.

If conversion failed, return default value without raising Exception.

Parameters
  • value (str) – Value to convert to int.

  • default (Optional[~T]) – Default value to use in case of failed conversion.

Return type

Union[int, ~T, None]