Api Reference

class alicorn.Alicorn(name='alicorn', config=None)

@service(self, cls)

A decorator for a Servicer class to add the service to the Alicorn instance. Automatically applies the injector decorator to all methods in the class.

Raises

TypeError – If the class does not inherit from a grpc servicer class

@dependency(self, f: Union[str, Callable])

Adds a named dependency for use with Depends

Examples

Create a named dependency called name.

>>> @dependency("name")
... def test():
...     pass

Create a named dependency called test2.

>>> @dependency
... def test2():
...     pass

@inject(self, f: Callable)

Inject dependencies into this method or function.

See also

alicorn.dependency.inject

@rpc_service(self, service_name: str = None)

Like the @service decorator but works on generic python classes. To define service methods use the @rpc_method_* decorators.

@rpc_method_unary(self, request: google.protobuf.message.Message, response: google.protobuf.message.Message, name: str = None)

Decorator for a unary-unary method

@rpc_method_unary_stream(self, request: google.protobuf.message.Message, response: google.protobuf.message.Message, name: str = None)

Decorator for a unary-stream method

@rpc_method_stream_unary(self, request: google.protobuf.message.Message, response: google.protobuf.message.Message, name: str = None)

Decorator for a stream-unary method

@rpc_method_stream(self, request: google.protobuf.message.Message, response: google.protobuf.message.Message, name: str = None)

Decorator for a stream-stream method

property debug

Debug mode

logger()

Return a cached logger instance

interceptor(f: Callable)

Adds an interceptor to the server

before_request(f: Callable[[RequestType, alicorn._ctx.GrpcContext, Callable], RequestType])

Register a callback that runs before a request is passed to the service method but after the request has been accepted. Called with the request message (or generator), context and a reference to the original method. The request can be re-written and returned.

This will run in the same thread as the service’s method.

If you are needing to run a callback before the request has been set-up then add middleware instead.

after_request(f: Callable[[alicorn._ctx.GrpcContext, Callable], None])

Register a callback that runs after a request has finished being processed by the service method but before the context has been torn down. Called with the context and a reference to the original method. As the request has already been served no response can be read or sent.

This will run in the same thread as the service’s method.

after_request_teardown(f: Callable)

Register a callback that runs after a request has been served and marked as complete. This is the equivalent of adding a done callback to the context. Does not receive any parameters.

This will run in the main thread.

add_extension(extension: alicorn.extension.Extension)

Adds an extension to this Alicorn instance

add_context_handler(handler: Callable[], Dict])

Add a handler that is called when an AlicornContext is created. The handler should return a dict that is added into the context during the request.

add_dependency(name: str, f: Callable)

Add a named dependency

register_signal(name: str) → alicorn.signal.Signal

Registers a signal that can be listened to and called. See alicorn.signal.Signal

unregister_signal(name: str)

Unregister a signal

Raises

AlicornSignalNotFound – If the signal does not exist

get_signal(name: str) → alicorn.signal.Signal

Return a registered Signal object

Raises

AlicornSignalNotFound – If the signal does not exist

register_signal_handler(name: str, f: Callable)

Registers the handler to the given signal name

unregister_signal_handler(name: str, f: Callable)

Removes the handler from the signal

get_handlers()

Return a dictionary of handlers for use with grpc-testing

stop(grace: Optional[int] = None)

Stop the server. This method returns immediately with a threading Event that indicates when all the requests have finished been processed

app_context()

Create an application context

auto_discover(name='service')

Automatically import modules that match the given base location and name

run(listen: Optional[Union[alicorn.port.SecurePort, alicorn.port.InsecurePort]] = None)

Start the grpc server. If no ports have been configured then use the host and port parameters to create an insecure port if specified. If not specified then raise an error.

Raises

AlicornConfigurationException – If no ports have been specified and listen is None

Discovery

alicorn.auto_discover(locations=None, name='service')

Automatically import modules that match the given base location and name

Extension

class alicorn.extension.Extension

An Alicorn extension.

Pass an instance of the extension to Alicorn before starting the server.

Example

>>> extension = Extension()
>>> alicorn.add_extension(extension)

abstract register(app: alicorn.Alicorn)

Called by the Alicorn instance when added as an extension

Dependencies

Alicorn dependency injector

alicorn.dependency.inject(app: alicorn.Alicorn = None)

A decorator to automatically inject dependencies into the wrapped function when called. If app is specified then allows named dependencies to be specified using Depends('name'), otherwise only callables can be evaluated

class alicorn.dependency.Depends(name_or_callable: Optional[Union[str, Callable]] = None)

Creates a new dependency-injected value that will be injected when called. If not initialised or initialised without any arguments it will call the type annotation of the argument. If a string is specified, it will find a reference to the value in the list of named dependencies created by alicorn.Alicorn.dependency. If a callable is specified, the callable will be called as the function is called.

Ports

class alicorn.port.InsecurePort(host: str, port: int)

Creates an insecure port for the grpc server.

Parameters

• host – The IP address or hostname to listen on

• port – The port to listen on

See also

grpc.Server.add_insecure_port

property host

Alias for field number 0

property port

Alias for field number 1

class alicorn.port.SecurePort(host: str, port: int, credentials: grpc.ServerCredentials)

Creates a secure port for the grpc server with the given credentials.

Parameters

• host – The IP address or hostname to listen on

• port – The port to listen on

• credentials – The grpc.ServerCredentials to secure the port with

See also

grpc.Server.add_secure_port

property host

Alias for field number 0

property port

Alias for field number 1

property credentials

Alias for field number 2

Signals

class alicorn.signal.Signal(name='Undefined')

A signal handler that manages callbacks registered against it. This works like a list but when called all items in the list are called with the given arguments. If a handler returns true then don’t call subsequent handlers

Example

>>> def handler(n):
...     print(n * 2)
>>> signal = Signal('test')
>>> signal.append(handler)
>>> signal(2)
4

Exceptions

exception alicorn.exception.AlicornException

A base exception for all Alicorn exceptions

exception alicorn.exception.AlicornConfigurationException

Raised on any configuration errors (eg: no ports specified for the server)

exception alicorn.exception.AlicornSignalNotFound

Raised when a signal name does not exist

exception alicorn.exception.AlicornStatusException(reason)

Base exception class for raising grpc Status Codes.

See also

grpc.StatusCode

exception alicorn.exception.GrpcStatusCancelled(reason)

exception alicorn.exception.GrpcStatusUnknown(reason)

exception alicorn.exception.GrpcStatusInvalidArgument(reason)

exception alicorn.exception.GrpcStatusDeadlineExceeded(reason)

exception alicorn.exception.GrpcStatusNotFound(reason)

exception alicorn.exception.GrpcStatusAlreadyExists(reason)

exception alicorn.exception.GrpcStatusPermissionDenied(reason)

exception alicorn.exception.GrpcStatusResourceExhausted(reason)

exception alicorn.exception.GrpcStatusFailedPrecondition(reason)

exception alicorn.exception.GrpcStatusAborted(reason)

exception alicorn.exception.GrpcStatusOutOfRange(reason)

exception alicorn.exception.GrpcStatusUnimplemented(reason)

exception alicorn.exception.GrpcStatusInternal(reason)

exception alicorn.exception.GrpcStatusDataLoss(reason)

exception alicorn.exception.GrpcStatusUnauthenticated(reason)