API Reference¶

`insanic.app`¶

class Insanic(name, router=None, error_handler=None, load_env=True, request_class=None, strict_slashes=True, log_config=None, configure_logging=True, *, app_config=(), version=None, initialize_insanic_listeners=True, initialize_insanic_middlewares=True, attach_monitor_endpoints=True)¶

Initialize Insanic Application.

Parameters

name (str) – Name of your service!
router (Optional[InsanicRouter]) – If you want to pass in your own router
error_handler (Optional[ErrorHandler]) – If you want to use your own ErrorHandler
load_env (bool) – If you want to load environment variables.
request_class (Optional[Request]) – The request class that insanic will use.
strict_slashes (bool) – Always append a “/” to the end of all routes.
log_config (Optional[dict]) – Your log configuration.
configure_logging (bool) – To configure logging
app_config (Iterable[Union[str, object]]) – order matters! Any configs later in the iterable will overwrite previous set settings
version (Optional[str]) – your application version!
initialize_insanic_listeners (bool) – Determines if you want to attach insanic’s default listeners.
initialize_insanic_middlewares (bool) – Determines if you want to attach insanic’s default middlewares.
attach_monitor_endpoints (bool) – Determines if you want to attach insanic’s default monitoring endpoints.

configure_version(version)¶

Configures the application version and sets the version on settings. This is especially necessary for microservices.

Precedence

version argument
APPLICATION_VERSION in settings
defaults to UNKNOWN

Parameters: version (str) – version of the service or application
Return type: None

initialize_listeners()¶

Initializes the default listeners for insanic.

Return type: None

initialize_middleware()¶

Initializes default middlewares for insanic.

Return type: None

verify_plugin_requirements()¶

Checks if the required plugins set in REQUIRED_PLUGINS have been installed and initialized.

Return type: None

plugin_initialized(plugin_name, instance)¶

Interface to attach plugin for plugin developers to use. After initialization of plugin, call this!

>>> app.plugin_initialized('name_of_plugin', self)

Parameters

plugin_name (str) – The name of your plugin. For plugin developers
instance (Any) – The initialized plugin instance.

Return type

None

run(host=None, port=None, *, debug=False, ssl=None, sock=None, workers=1, protocol=<class 'insanic.protocol.InsanicHttpProtocol'>, backlog=65535, stop_event=None, register_sys_signals=True, access_log=True, auto_reload=None, unix=None, loop=None, **kwargs)¶

Run the HTTP Server and listen until keyboard interrupt or term signal. On termination, drain connections before closing.

Insanic overrides this because we want to use InsanicHttpProtocol instead of Sanic’s HttpProtocol.

Parameters

host (str) – Address to host on
port (int) – Port to host on
debug (bool) – Enables debug output (slows server)
auto_reload (Optional[bool]) – Reload app whenever its source code is changed. Enabled by default in debug mode.
ssl (SSLContext or dict) – SSLContext, or location of certificate and key for SSL encryption of worker(s)
sock (socket) – Socket for the server to accept connections from
workers (int) – Number of processes received before it is respected
protocol (type[Protocol]) – Subclass of asyncio Protocol class
backlog (int) – a number of unaccepted connections that the system will allow before refusing new connections
stop_event (None) – event to be triggered before stopping the app - deprecated
register_sys_signals (bool) – Register SIG* events
access_log (bool) – Enables writing access logs (slows server)
unix (str) – Unix socket to listen on instead of TCP port
loop (None) –

Returns

Nothing

`insanic.conf`¶

class InsanicConfig(settings_module='', defaults=None, load_env=True, keep_alive=None)¶

Bases: sanic.config.Config

Load order:

Sanic DEFAULT_CONFIG
defaults sent in through argument
SANIC_PREFIX environment variables
Insanic insanic.conf.global_settings
Service configs loaded from INSANIC_SETTINGS_MODULE env variable
INSANIC_PREFIX environment variables

Sequentially loaded variables overwrite settings declared in previous steps.

load_from_service()¶

Loads a config from defined settings module.

If the module doesn’t exist raise exception

Return type: None

`insanic.exceptions`¶

exception ImproperlyConfigured¶

Bases: Exception

Insanic is somehow improperly configured

exception APIException(description=None, *, error_code=None, status_code=None)¶

Bases: Exception

Base class for REST framework exceptions. Subclasses should provide .status_code, .error_code and .message properties.

Parameters

description (Optional[str]) – A description of the error.
error_code (Optional[Enum]) – The error code associated with the exception.
status_code (Optional[int]) – The status code to set for this exception.

message = 'An unknown error occurred'¶

i18n = False¶

status_code = 500¶

error_code = 999999¶

exception ParseError(description=None, *, error_code=None, status_code=None)¶

Bases: insanic.exceptions.APIException

Parameters

description (Optional[str]) –
error_code (Optional[enum.Enum]) –
status_code (Optional[int]) –

status_code = 400¶

message = 'Malformed request.'¶

exception BadRequest(description=None, *, error_code=None, status_code=None)¶

Bases: insanic.exceptions.APIException

Parameters

description (Optional[str]) –
error_code (Optional[enum.Enum]) –
status_code (Optional[int]) –

status_code = 400¶

message = 'Bad request.'¶

exception InvalidUsage(description=None, *, error_code=None, status_code=None)¶

Bases: insanic.exceptions.APIException

Parameters

description (Optional[str]) –
error_code (Optional[enum.Enum]) –
status_code (Optional[int]) –

status_code = 400¶

message = 'Invalid Usage'¶

exception ValidationError(description=None, *, error_code=None, status_code=None)¶

Bases: insanic.exceptions.APIException

Parameters

description (Optional[str]) –
error_code (Optional[enum.Enum]) –
status_code (Optional[int]) –

status_code = 400¶

message = 'Validation Error'¶

exception NotFound(description=None, *, error_code=None, status_code=None)¶

Bases: insanic.exceptions.APIException

Parameters

description (Optional[str]) –
error_code (Optional[enum.Enum]) –
status_code (Optional[int]) –

status_code = 404¶

message = 'Not found.'¶

exception AuthenticationFailed(description=None, *, error_code=None, status_code=None)¶

Bases: insanic.exceptions.APIException

Parameters

description (Optional[str]) –
error_code (Optional[enum.Enum]) –
status_code (Optional[int]) –

status_code = 401¶

message = 'Incorrect authentication credentials.'¶

exception ServiceAuthenticationFailed(description=None, *, error_code=None, status_code=None)¶

Bases: insanic.exceptions.APIException

Parameters

description (Optional[str]) –
error_code (Optional[enum.Enum]) –
status_code (Optional[int]) –

status_code = 401¶

message = 'Incorrect authentication credentials from service.'¶

exception NotAuthenticated(description=None, *, error_code=None, status_code=None)¶

Bases: insanic.exceptions.APIException

Parameters

description (Optional[str]) –
error_code (Optional[enum.Enum]) –
status_code (Optional[int]) –

status_code = 401¶

message = 'Authentication credentials were not provided.'¶

exception PermissionDenied(description=None, *, error_code=None, status_code=None)¶

Bases: insanic.exceptions.APIException

Parameters

description (Optional[str]) –
error_code (Optional[enum.Enum]) –
status_code (Optional[int]) –

status_code = 403¶

message = 'You do not have permission to perform this action.'¶

exception MethodNotAllowed(method, description=None)¶

Bases: insanic.exceptions.APIException

status_code = 405¶

message = "Method '%s' not allowed."¶

error_code = 999501¶

exception NotAcceptable(description=None, *, error_code=None, status_code=None)¶

Bases: insanic.exceptions.APIException

Parameters

description (Optional[str]) –
error_code (Optional[enum.Enum]) –
status_code (Optional[int]) –

status_code = 406¶

message = 'Could not satisfy the request Accept header'¶

exception UnsupportedMediaType(media_type, description=None)¶

Bases: insanic.exceptions.APIException

status_code = 415¶

message = 'Unsupported media type.'¶

description = "Unsupported media type '%s' in request."¶

exception Throttled(wait=None, description=None)¶

Bases: insanic.exceptions.APIException

status_code = 429¶

message = 'Request was throttled.'¶

extra_detail = ('Expected available in %(wait)d second.', 'Expected available in %(wait)d seconds.')¶

error_code = 999502¶

exception FieldError¶

Bases: Exception

Some kind of problem with a model field.

exception RawPostDataException¶

Bases: Exception

You cannot access raw_post_data from a request that has multipart/* POST data if it has been accessed via POST, FILES, etc..

exception ServiceUnavailable503Error(description=None, *, error_code=None, status_code=None)¶

Bases: insanic.exceptions.APIException

Parameters

description (Optional[str]) –
error_code (Optional[enum.Enum]) –
status_code (Optional[int]) –

status_code = 503¶

message = 'Service unavailable.'¶

exception ResponseTimeoutError(description=None, *, error_code=None, status_code=None)¶

Bases: insanic.exceptions.APIException

Parameters

description (Optional[str]) –
error_code (Optional[enum.Enum]) –
status_code (Optional[int]) –

status_code = 503¶

error_code = 999604¶

message = 'Response timeout.'¶

exception RequestTimeoutError(description=None, *, error_code=None, status_code=None)¶

Bases: insanic.exceptions.APIException

Parameters

description (Optional[str]) –
error_code (Optional[enum.Enum]) –
status_code (Optional[int]) –

status_code = 408¶

message = 'Request timeout.'¶

exception UnprocessableEntity422Error(description=None, *, error_code=None, status_code=None)¶

Bases: insanic.exceptions.APIException

Parameters

description (Optional[str]) –
error_code (Optional[enum.Enum]) –
status_code (Optional[int]) –

status_code = 422¶

message = 'Unprocessable Entity'¶

exception SanicInvalidUsage(description=None, status_code=None)¶: Bases: insanic.exceptions.InvalidUsage

exception SanicNotFound(description=None, *, error_code=None, status_code=None)¶

Bases: insanic.exceptions.APIException

Parameters

description (Optional[str]) –
error_code (Optional[enum.Enum]) –
status_code (Optional[int]) –

status_code = 404¶

message = 'Not Found'¶

error_code = 999404¶

exception SanicMethodNotSupported(description=None, *, error_code=None, status_code=None)¶

Bases: insanic.exceptions.APIException

Parameters

description (Optional[str]) –
error_code (Optional[enum.Enum]) –
status_code (Optional[int]) –

status_code = 405¶

message = 'Method not allowed.'¶

error_code = 999501¶

`insanic.listeners`¶

before_server_start_verify_plugins(app, loop, **kwargs)¶: Check if all the required plugins have been initialized.

before_server_start_set_task_factory(app, loop, **kwargs)¶: Sets the task factory to pass context.

async after_server_start_connect_database(app, loop=None, **kwargs)¶: Sets the connections object to the running loop.

async after_server_stop_clean_up(app, loop, **kwargs)¶: Clean up all connections and close service client connections.

`insanic.loading`¶

get_service(service_name)¶

Helper function to get the service connection object

Parameters: service_name (str) – Name of the service defined in settings.
Return type: Service

`insanic.middleware`¶

request_middleware(request)¶

Request middleware that runs on all requests. Tracks the request count and sets a correlation id to the asyncio task if included in the headers.

Parameters: request – The Request object.
Return type: None

`insanic.permissions`¶

Basic Permissions provided by Insanic.

class BasePermission¶

Bases: object

A base class from which all permission classes should inherit.

class AllowAny¶

Bases: insanic.permissions.BasePermission

Allow any access. This isn’t strictly required, since you could use an empty permission_classes list, but it’s useful because it makes the intention more explicit.

class IsAuthenticated¶

Bases: insanic.permissions.BasePermission

Allows access only to authenticated users.

class IsAdminUser¶

Bases: insanic.permissions.BasePermission

Allows access only to admin users.

class IsAuthenticatedOrReadOnly¶

Bases: insanic.permissions.BasePermission

The request is authenticated as a user, or is a read-only request.

class IsOwnerOrAdmin¶

Bases: insanic.permissions.BasePermission

Custom permission to only allow owners of an object to view or edit it.

class IsAnonymousUser¶

Bases: insanic.permissions.BasePermission

Permission to check this api can only be access by non authenticated user.

class IsServiceOnly¶

Bases: insanic.permissions.BasePermission

Permission to check this api can only be access by another service

`insanic.request`¶

class Request(url_bytes, headers, version, method, transport, app, authenticators=None)¶

Bases: sanic.request.Request

property id¶

Gets a unique request id from either the header with REQUEST_ID_HEADER_FIELD config, or generates it’s own unique request id.

Return type: str

property query_params¶

More semantically correct name for request.GET.

Return type: dict

property user¶

Returns the user associated with the current request, as authenticated by the authentication classes provided to the request.

Return type: User

property service¶

Returns the service associated with the current request, as authenticated by the authentication classes provided to the request.

Return type: RequestService

property auth¶

Returns any non-user authentication information associated with the request, such as an authentication token.

Return type: BaseAuthentication

property successful_authenticator¶

Return the instance of the authentication instance class that was used to authenticate the request, or None.

Return type: BaseAuthentication

property data¶

A single interface for getting the body of the request without needing to know the content type of the request. From django-rest-framework.

Return type: RequestParameters

`insanic.router`¶

class InsanicRouter¶

Bases: sanic.router.Router

property routes_public¶

Gathers all the registered routes and determines if they have been decorated with the public_facing decorator.

Return type: dict

`insanic.models`¶

class User(*, id='', level=- 1, is_authenticated=False, **kwargs)¶

Bases: object

Basic User model used for Insanic’s Authentication

Parameters

id (str) – The user id.
level (int) – The user’s level (for determining if banned, is staff).
is_authenticated (Union[bool, int]) – If the user is authenticated.

id¶

level¶

property is_staff¶

If user’s is determined to be a staff member

Return type: bool

property is_authenticated¶: If user is resolved to be authenticated :rtype: int :return:

property is_active¶

If user is an active user.

Return type: bool

property is_banned¶

If user is a banned user.

Return type: bool

class RequestService(*, source, aud, source_ip, is_authenticated)¶

Bases: object

Basic Request Service model.

Parameters

source (str) – The Request service’s name.
aud (str) – The current application’s name.
source_ip (str) – The ip of the requested service.
is_authenticated (Union[int, bool]) – If the service is authenticated.

request_service¶

destination_service¶

source_ip¶

is_authenticated¶

property is_valid¶

If it is a valid request to the current application.

Return type: bool

AnonymousUser = <insanic.models._AnonymousUser object>¶: An AnonmyousUser instance.

AnonymousRequestService = <insanic.models.RequestService object>¶: An Anonymous Request Service instance.

to_header_value(user)¶

A helper method to convert the User object to str.

Parameters: user (User) –
Return type: str
Returns: A serialized User string.

`insanic.services`¶

class Service(service_name, partial_path=None)¶

The service object to facilitate sending requests to other services.

Parameters

service_name (str) – The name of the service to send a request to.
partial_path (Optional[str]) – Base path of the endpoint to send to.

property client¶

The httpx.AsyncClient that will be used to send async requests.

Return type: AsyncClient

property host¶

The host portion of the url.

Return type: str

property port¶

The port portion of the url.

Return type: int

async close_client()¶

Close the async client on shutdown.

Return type: None

http_dispatch(method, endpoint, *, query_params=None, payload=None, files=None, headers=None, propagate_error=False, include_status_code=False, response_timeout=<httpx._config.UnsetType object>, retry_count=None, **kwargs)¶

Interface for sending requests to other services.

Parameters

method (str) – method to send request (GET, POST, PATCH, PUT, etc)
endpoint (str) – the path to send request to (eg /api/v1/..)
query_params (Optional[dict]) – query params to attach to url
payload (Optional[dict]) – the data to send on any non GET requests
files (Optional[dict]) – if any files to send with request, must be included here
headers (Optional[dict]) – headers to send along with request
propagate_error (bool) – if you want to raise on 400 or greater status codes
include_status_code (bool) – if you want this method to return the response with the status code
response_timeout (int) – if you want to increase the timeout for this requests
retry_count (Optional[int]) – number times you want to retry the request if failed on server errors

`insanic.throttles`¶

Default throttles provided by Insanic.

class BaseThrottle¶

Bases: object

Rate throttling of requests.

async allow_request(request, view)¶

Return True if the request should be allowed, False otherwise.

Return type: bool

wait()¶

Optionally, return a recommended number of seconds to wait before the next request.

Return type: int

class SimpleRateThrottle¶

Bases: insanic.throttles.BaseThrottle

A simple cache implementation, that only requires .get_cache_key() to be overridden. The rate (requests / seconds) is set by a rate attribute on the View class. The attribute is a string of the form ‘number_of_requests/period’. Period should be one of: (‘s’, ‘sec’, ‘m’, ‘min’, ‘h’, ‘hour’, ‘d’, ‘day’) Previous request information used for throttling is stored in the cache.

timer()¶

time() -> floating point number

Return the current time in seconds since the Epoch. Fractions of a second may be present if the system clock provides them.

async get_cache_key(request, view)¶

Should return a unique cache-key which can be used for throttling. Must be overridden. May return None if the request should not be throttled.

Return type

str

Parameters

request (sanic.request.Request) –
view (sanic.views.HTTPMethodView) –

get_rate()¶

Determine the string representation of the allowed request rate.

Return type: str

parse_rate(rate)¶

Given the request rate string, return a two tuple of: <allowed number of requests>, <period of time in seconds>

Return type: tuple
Parameters: rate (str) –

async allow_request(request, view)¶

Implement the check to see if the request should be throttled. On success calls throttle_success. On failure calls throttle_failure.

Return type

bool

Parameters

request (sanic.request.Request) –
view (sanic.views.HTTPMethodView) –

async throttle_success()¶

Inserts the current request’s timestamp along with the key into the cache.

Return type: bool

throttle_failure()¶

Called when a request to the API has failed due to throttling.

Return type: bool

wait()¶

Returns the recommended next request time in seconds.

Return type: int

class AnonRateThrottle¶

Bases: insanic.throttles.SimpleRateThrottle

Limits the rate of API calls that may be made by a anonymous users. The IP address of the request will be used as the unique cache key.

async get_cache_key(request, view)¶

Should return a unique cache-key which can be used for throttling. Must be overridden. May return None if the request should not be throttled.

Return type

str

Parameters

request (sanic.request.Request) –
view (sanic.views.HTTPMethodView) –

class UserRateThrottle¶

Bases: insanic.throttles.SimpleRateThrottle

Limits the rate of API calls that may be made by a given user. The user id will be used as a unique cache key if the user is authenticated. For anonymous requests, the IP address of the request will be used.

async get_cache_key(request, view)¶

Should return a unique cache-key which can be used for throttling. Must be overridden. May return None if the request should not be throttled.

Return type

str

Parameters

request (sanic.request.Request) –
view (sanic.views.HTTPMethodView) –

class ScopedRateThrottle¶

Bases: insanic.throttles.SimpleRateThrottle

Limits the rate of API calls by different amounts for various parts of the API. Any view that has the throttle_scope property set will be throttled. The unique cache key will be generated by concatenating the user id of the request, and the scope of the view being accessed.

async allow_request(request, view)¶

Implement the check to see if the request should be throttled. On success calls throttle_success. On failure calls throttle_failure.

Return type

bool

Parameters

request (sanic.request.Request) –
view (sanic.views.HTTPMethodView) –

async get_cache_key(request, view)¶

If view.throttle_scope is not set, don’t apply this throttle. Otherwise generate the unique cache key by concatenating the user id with the ‘.throttle_scope` property of the view.

Return type

str

Parameters

request (sanic.request.Request) –
view (sanic.views.HTTPMethodView) –

`insanic.utils.datetime`¶

get_utc_timestamp()¶

Returns the current utc timestamp with decimals.

>>> get_utc_datetime()
1531279648.991617

Return type: float

get_utc_datetime()¶

Returns the current utc datetime object.

Return type: datetime

timestamp_to_datetime(timestamp=None, units=None)¶

Converts a timestamp to datetime. Assumes timestamp is in utc timezone. If not passed gets the current timestamp and converts that.

If passing in timestamp, must supply units(either ms or s) to depending on the units of the timestamp provided.

Parameters

timestamp (Optional[int]) – Defaults to utc timestamp
units (Optional[str]) – either “ms” or “s” if timestamp is passed

Return type

datetime

timestamp_seconds_to_datetime(timestamp)¶

Wrapper for timestamp_to_datetime

Parameters: timestamp (Union[int, float]) – A timestamp in seconds
Return type: datetime

timestamp_milliseconds_to_datetime(timestamp)¶

Wrapper for timestamp_to_datetime

Parameters: timestamp (Union[int, float]) – A timestamp in milliseconds
Return type: datetime

timestamp_to_iso(timestamp=None, units=None)¶

Takes a timestamp and converts it to a iso formatted string

Parameters

timestamp (Union[int, float, None]) –
units (Optional[str]) – either “ms” or “s”

Return type

str

iso_to_datetime(datetime_string)¶

Takes an iso formatted datetime string and tries to convert it to a datetime object

Parameters: datetime_string (str) – An iso formatted datetime string
Return type: datetime

iso_to_timestamp(iso)¶

Takes an iso formatted string and tries to convert it to a timestamp

Parameters: iso (str) – An iso formatted datetime string
Return type: float

`insanic.views`¶

class InsanicView¶

Bases: sanic.views.HTTPMethodView

property allowed_methods¶: Wrap Django’s private _allowed_methods interface in a public property.

perform_authentication(request)¶

Perform authentication on the incoming request.

Note that if you override this and simply ‘pass’, then authentication will instead be performed lazily, the first time either request.user or request.auth is accessed.

check_permissions(request)¶: Check if the request should be permitted. Raises an appropriate exception if the request is not permitted.

permission_denied(request, message=None)¶: If request is not permitted, determine what kind of exception to raise.

get_permissions()¶: Instantiates and returns the list of permissions that this view requires.

get_throttles()¶: Instantiates and returns the list of throttles that this view uses.

throttled(request, wait)¶: If request is throttled, determine what kind of exception to raise.

async check_throttles(request)¶: Check if request should be throttled. Raises an appropriate exception if the request is throttled.

get_authenticators()¶: Instantiates and returns the list of authenticators that this view can use.

async prepare_http(request, *args, **kwargs)¶: .dispatch() is pretty much the same as Django’s regular dispatch, but with extra hooks for startup, finalize, and exception handling.

async dispatch_request(request, *args, **kwargs)¶: .dispatch() is pretty much the same as Django’s regular dispatch, but with extra hooks for startup, finalize, and exception handling.

API Reference¶

insanic.app¶

insanic.conf¶

insanic.exceptions¶

insanic.listeners¶

insanic.loading¶

insanic.middleware¶

insanic.permissions¶

insanic.request¶

insanic.router¶

insanic.models¶

insanic.services¶

insanic.throttles¶

insanic.utils.datetime¶

insanic.views¶

`insanic.app`¶

`insanic.conf`¶

`insanic.exceptions`¶

`insanic.listeners`¶

`insanic.loading`¶

`insanic.middleware`¶

`insanic.permissions`¶

`insanic.request`¶

`insanic.router`¶

`insanic.models`¶

`insanic.services`¶

`insanic.throttles`¶

`insanic.utils.datetime`¶

`insanic.views`¶