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=None, *operations, schema=None, spec=None, 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, schema_loader=None, cache_create_schema_and_spec=False, validate_email_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 using schema first approach and 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: :rtype: Application

  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 how 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 stdout. 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().

To simplify things rororo expects on OpenAPI 3 path and do reading schema from file and specifying openapi_core.schema.specs.models.Spec instance inside of rororo.openapi.setup_openapi() call.

However, it is possible to completely customize this default behaviour and pass OpenAPI schema and spec instance directly. In that case schema keyword argument should contains raw OpenAPI 3 schema as Dict[str, Any], while spec to be an openapi_core.schema.specs.models.Spec instance.

This behaviour might be helpful if you’d like to cache reading schema and instantiating spec within tests or other environments, which requires multiple rororo.openapi.setup_openapi() calls.

from pathlib import Path

import yaml
from aiohttp import web
from openapi_core.shortcuts import create_spec
from rororo import setup_openapi


# Reusable OpenAPI data
openapi_yaml = Path(__file__).parent / "openapi.yaml"
schema = yaml.load(
    openapi_yaml.read_bytes(), Loader=yaml.CSafeLoader
)
spec = create_spec(schema)

# Create OpenAPI 3 aiohttp.web server application
app = setup_openapi(web.Application(), schema=schema, spec=spec)

For default behaviour, with passing schema_path, there are few options on customizing schema load process as well,

By default, rororo will use json.loads() to load OpenAPI schema content from JSON file and yaml.CSafeLoader if it is available to load schema content from YAML files (with fallback to yaml.SafeLoader). But, for performance considreations, you might use any other function to load the schema. Example below illustrates how to use ujson.loads function to load content from JSON schema,

import ujson

app = setup_openapi(
    web.Application(),
    Path(__file__).parent / "openapi.json",
    operations,
    schema_loader=ujson.loads,
)

Schema loader function expects bytes as only argument and should return Dict[str, Any] as OpenAPI schema dict.

Danger

By default rororo does not cache slow calls to read OpenAPI schema and creating its spec. But sometimes, for example in tests, it is sufficient to cache those calls. To enable cache behaviour pass cache_create_schema_and_spec=True or even better, cache_create_schema_and_spec=settings.is_test.

But this may result in unexpected issues, as schema and spec will be cached once and on next call it will result cached data instead to attempt read fresh schema from the disk and instantiate OpenAPI Spec instance.

By default, rororo using validate_email function from email-validator library to validate email strings, which has been declared in OpenAPI schema as,

components:
  schemas:
    Email:
      type: "string"
      format: "email"

In most cases validate_email(email) call should be enough, but in case if you need to pass extra **kwargs for validating email strings, setup validate_email_kwargs such as,

app = setup_openapi(
    web.Application(),
    Path(__file__).parent / "openapi.json",
    operations,
    validate_email_kwargs={"check_deliverability": False},
)
class rororo.openapi.OperationTableDef(handlers=_Nothing.NOTHING, views=_Nothing.NOTHING)[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:
    ...

Class based views supported as well. In most generic way you just need to decorate your view with @operations.register decorator and ensure that operationId equals to view method qualified name (__qualname__).

For example,

@operations.register
class UserView(web.View):
    async def get(self) -> web.Response:
        ...

expects for operation ID UserView.get to be declared in OpenAPI schema.

In same time,

@operations.register("users")
class UserView(web.View):
    async def get(self) -> web.Response:
        ...

expects for operation ID users.get to be declared in OpenAPI schema.

Finally,

@operations.register
class UserView(web.View):
    @operations.register("me")
    async def get(self) -> web.Response:
        ...

expects for operation ID me to be declared in OpenAPI schema.

When the class based view provides mutliple view methods (for example delete, get, patch & put) rororo expects that OpenAPI schema contains operation IDs for each of view method.

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

rororo.openapi.read_openapi_schema(path, *, loader=None)[source]#

Read OpenAPI Schema from given path.

By default, when loader is not explicitly passed, attempt to guess schema loader function from path extension.

loader should be a callable, which receives bytes and returns Dict[str, Any] of OpenAPI Schema.

By default, next schema loader used, :rtype: Dict[str, Any]

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

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

rororo.openapi.data#

Provide structures for OpenAPI data.

class rororo.openapi.data.OpenAPIContext(request, app, config_dict, parameters=_Nothing.NOTHING, security=_Nothing.NOTHING, data=None)[source]#

All data associated with current request to OpenAPI handler.

Contains only valid parameters, security data & request body data. Example bellow illustrates how to work with context data,

from rororo import get_openapi_context
from rororo.openapi.exceptions import InvalidCredentials


async def create_user(request: web.Request) -> web.Response:
    context = get_openapi_context(request)

    # Authenticate current user (accessing security data)
    if not authenticate(api_key=context.security["apiKey"]):
        raise InvalidCredentials()

    # Add new user (accessing request body data)
    async with request.config_dict["db"].acquire() as conn:
        user = await create_user(
            conn,
            email=context.data["email"],
            password=context.data["password"],
        )

    # Return response due to query string param
    # (accessing parameters data)
    if context.parameters.query["login"]:
        return web.json_response(
            request.app.router["login"].url_for()
        )

    return web.json_response(user.to_api_dict())

rororo.openapi.exceptions#

class rororo.openapi.BadRequest(message=None, *, headers=None)[source]#

Bad request basic error.

class rororo.openapi.SecurityError(message=None, *, headers=None)[source]#

Request is not secured, but should.

class rororo.openapi.BasicSecurityError(message=None, *, headers=None)[source]#

Basic authentication is not provided.

class rororo.openapi.InvalidCredentials(message=None, *, headers=None)[source]#

Invalid credentials provided for authentication.

class rororo.openapi.BasicInvalidCredentials(message=None, *, headers=None)[source]#

Invalid credentials provided for basic authentication.

class rororo.openapi.ObjectDoesNotExist(label='Object', *, message=None, headers=None)[source]#

Object does not exist basic error.

class rororo.openapi.ValidationError(*, message=None, errors=None)[source]#

Request / response validation error.

There is an ability to wrap openapi-core request / response validation errors via from_request_errors() class method as in same time to create a validation error from the dict, like:

raise ValidationError.from_dict(
    body={"name": "Name is not unique"}
)
raise ValidationError.from_dict(
    parameters={"X-Api-Key": "Outdated API key"}
)
raise ValidationError.from_dict(
    body={
        0: {
            "data": {
                0: {"name": "Field required"},
                1: {"description": "Field required"},
            },
        },
    },
)

Given interface is recommended for end users, who want to raise a custom validation errors within their operation handlers.

class rororo.openapi.ServerError(message=None, *, headers=None)[source]#

Server error.

rororo.openapi.validation_error_context(*path)[source]#

Context manager to specify the proper path for validation error item.

Main purpose to setup validation error path for external validators.

For example, when you need to have reusable validator for phone number, you can organize your code as follows, :rtype: Iterator[Tuple[Union[int, str], ...]]

  1. Create validate_phone function, which will check whether string is a valid phone number. If not raise a rororo.openapi.ValidationError with specific message.

  2. Reuse validate_phone function anywhere in your code by wrapping call into validation_error_context context manager

First level errors,

@operations.register
async def create_user(request: web.Request) -> web.Response:
    data = get_validated_data(request)

    with validation_error_context("body", "phone"):
        phone = validate_phone(data["phone"])

    ...

Secon level errors,

@operations.register
async def create_order(request: web.Request) -> web.Response:
    with validation_error_context("body"):
        order = validate_order(get_validated_data(request))

    ...


def validate_order(data: MappingStrAny) -> Order:
    with validation_error_context("user", "phone"):
        user_phone = validate_phone(data["user"]["phone"])
    ...
rororo.openapi.get_current_validation_error_loc()[source]#

Return current validation error location (path).

In most cases you don’t need to call this function directly, but it might be useful for logging purposes.

Return type:

Tuple[Union[int, str], ...]

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='localhost', port=8080, debug=False, level='dev', time_zone='UTC', first_weekday=0, locale='en_US.UTF-8', sentry_dsn=None, sentry_release=None)[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 environ
from rororo.settings import BaseSettings


@environ.config(prefix="", frozen=True)
class Settings(BaseSettings):
    other_name: str = environ.var(
        name="OTHER_NAME", default="other-value"
    )
rororo.settings.setup_settings_from_environ(app, settings_class, *, environ=None, loggers=None, remove_root_handlers=False)[source]#

Shortcut for instantiating settings from environ and applying them for given aiohttp.web app.

This function calls settings_class.from_environ() method for you.

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

Return type:

Application

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

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

  • **optionals (Any) –

    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, ModuleType, 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[ModuleType, 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 4.0.

Parameters:
  • key (str) – Environment var key.

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

Return type:

Union[str, TypeVar(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 (str) – Enable logging for each logger in sequence.

  • **kwargs (Any) – 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 4.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 (str) – Use Sentry logging for each logger in the sequence. If the sequence is empty use Sentry logging to each available logger.

  • **kwargs (Any) – 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))

Deprecated since version 3.0.2: As aioredis library deprecated in favor of redis library, which supports instantiating client instance from URL string - this function will be removed in 4.0.

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) – 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[TypeVar(T)]) – Default value to use in case of failed conversion.

Return type:

Union[int, TypeVar(T), None]