PySOA API Reference Documentation

This file contains the reference documentation for the classes and functions you are likely to consume when using PySOA. It does not contain the reference documentation for all code in PySOA, which you can view by browsing the source code.

class pysoa.client.client.Client(config, expansion_config=None, settings_class=None, context=None)

Bases: object

The Client provides a simple interface for calling actions on services and supports both sequential and parallel action invocation.

exception CallActionError(actions=None)

Bases: pysoa.client.errors.PySOAClientError

Raised by Client.call_*** methods when a job response contains one or more action errors. Stores a list of ActionResponse objects and has a string representation cleanly displaying the actions’ errors.

__init__(actions: Union[List[pysoa.common.types.ActionResponse], None] = None) → None

Parameters

actions – The list of all actions that have errors (not actions without errors), available as an actions property on the exception instance.

__str__()

Return str(self).

exception CallJobError(errors=None)

Bases: pysoa.client.errors.PySOAClientError

Raised by Client.call_*** methods when a job response contains one or more job errors. Stores a list of Error objects and has a string representation cleanly displaying the errors.

__init__(errors: Union[List[pysoa.common.errors.Error], None] = None) → None

Parameters

errors – The list of all errors in this job, available as an errors property on the exception instance.

__str__()

Return str(self).

FutureResponse

alias of FutureSOAResponse

exception ImproperlyConfigured

Bases: pysoa.client.errors.PySOAClientError

Raised when this client is improperly configured to call the specified service.

exception InvalidExpansionKey

Bases: pysoa.client.errors.PySOAClientError

Raised when this client is improperly configured to perform the specified expansion.

JobError

alias of pysoa.client.errors.CallJobError

__init__(config: Mapping[str, Mapping[str, Any]], expansion_config: Union[Mapping[str, Any], None] = None, settings_class: Union[Type[pysoa.client.settings.ClientSettings], None] = None, context: Union[Dict[str, Any], None] = None) → None

Parameters

• config – The entire client configuration dict, whose keys are service names and values are settings dicts abiding by the pysoa.client.settings.ClientSettings schema

• expansion_config – The optional expansion configuration dict, if this client supports expansions, which is a dict abiding by the pysoa.client.expander.ExpansionSettings schema

• settings_class – An optional settings schema enforcement class to use, which overrides the default of pysoa.client.settings.ClientSettings

• context – An optional base request context that will be used for all requests this client instance sends (individual calls can add to and override the values supplied in this context dict)

call_action(service_name: str, action: str, body: Dict[str, Any] = None, expansions: Dict[str, List[str]] = None, raise_job_errors: bool = True, raise_action_errors: bool = True, timeout: Optional[int] = None, switches: Union[List[int], AbstractSet[int], None] = None, correlation_id: Optional[str] = None, context: Union[Dict[str, Any], None] = None, control_extra: Union[Dict[str, Any], None] = None) → <class 'pysoa.common.types.ActionResponse'>

Build and send a single job request with one action.

Returns the action response or raises an exception if the action response is an error (unless raise_action_errors is passed as False) or if the job response is an error (unless raise_job_errors is passed as False).

This method performs expansions if the Client is configured with an expansion converter.

Parameters

• service_name – The name of the service to call.

• action – The name of the action to call.

• body – The action request body.

• expansions – A dictionary representing the expansions to perform.

• raise_job_errors – Whether to raise a pysoa.client.errors.CallJobError if the job response contains errors (defaults to True).

• raise_action_errors – Whether to raise a pysoa.client.errors.CallActionError if any action responses contain errors (defaults to True).

• timeout – If provided, this will override the default transport timeout values to; requests will expire after this number of seconds plus some buffer defined by the transport, and the client will not block waiting for a response for longer than this amount of time.

• switches – A list of switch value integers.

• correlation_id – The request correlation ID.

• context – A dictionary of extra values to include in the context header.

• control_extra – A dictionary of extra values to include in the control header.

Returns

The action response.

Raises

pysoa.common.transport.errors.PySOATransportError, pysoa.client.errors.CallActionError, pysoa.client.errors.CallJobError

call_action_future(service_name: str, action: str, body: Union[Dict[str, Any], None] = None, expansions: Dict[str, List[str]] = None, raise_job_errors: bool = True, raise_action_errors: bool = True, timeout: Optional[int] = None, switches: Union[List[int], AbstractSet[int], None] = None, correlation_id: Optional[str] = None, context: Union[Dict[str, Any], None] = None, control_extra: Union[Dict[str, Any], None] = None) → pysoa.client.client.FutureSOAResponse[pysoa.common.types.ActionResponse]

This method is identical in signature and behavior to call_action(), except that it sends the request and then immediately returns a FutureResponse instead of blocking waiting on a response and returning an pysoa.common.types.ActionResponse. Just call result(timeout=None) on the future response to block for an available response. Some of the possible exceptions may be raised when this method is called; others may be raised when the future is used.

Returns

A future from which the action response can later be retrieved

Raises

pysoa.common.transport.errors.PySOATransportError

call_actions(service_name: str, actions: Union[List[pysoa.common.types.ActionRequest], List[Dict[str, Any]]], expansions: Dict[str, List[str]] = None, raise_job_errors: bool = True, raise_action_errors: bool = True, timeout: Optional[int] = None, switches: Union[List[int], AbstractSet[int], None] = None, correlation_id: Optional[str] = None, continue_on_error: bool = False, context: Union[Dict[str, Any], None] = None, control_extra: Union[Dict[str, Any], None] = None) → <class 'pysoa.common.types.JobResponse'>

Build and send a single job request with one or more actions.

Returns a list of action responses, one for each action in the same order as provided, or raises an exception if any action response is an error (unless raise_action_errors is passed as False) or if the job response is an error (unless raise_job_errors is passed as False).

This method performs expansions if the Client is configured with an expansion converter.

Parameters

• service_name – The name of the service to call.

• actions – A list of pysoa.common.types.ActionRequest objects and/or dicts that can be converted to ActionRequest objects.

• expansions – A dictionary representing the expansions to perform.

• raise_job_errors – Whether to raise a pysoa.client.errors.CallJobError if the job response contains errors (defaults to True).

• raise_action_errors – Whether to raise a pysoa.client.errors.CallActionError if any action responses contain errors (defaults to True).

• timeout – If provided, this will override the default transport timeout values to; requests will expire after this number of seconds plus some buffer defined by the transport, and the client will not block waiting for a response for longer than this amount of time.

• switches – A list of switch value integers.

• correlation_id – The request correlation ID.

• continue_on_error – Whether the service should continue executing further actions once one action has returned errors.

• context – A dictionary of extra values to include in the context header.

• control_extra – A dictionary of extra values to include in the control header.

Returns

The job response.

Raises

pysoa.common.transport.errors.PySOATransportError, pysoa.client.errors.CallActionError, pysoa.client.errors.CallJobError

call_actions_future(service_name: str, actions: Union[List[pysoa.common.types.ActionRequest], List[Dict[str, Any]]], expansions: Dict[str, List[str]] = None, raise_job_errors: bool = True, raise_action_errors: bool = True, timeout: Optional[int] = None, switches: Union[List[int], AbstractSet[int], None] = None, correlation_id: Optional[str] = None, continue_on_error: bool = False, context: Union[Dict[str, Any], None] = None, control_extra: Union[Dict[str, Any], None] = None) → pysoa.client.client.FutureSOAResponse[pysoa.common.types.JobResponse]

This method is identical in signature and behavior to call_actions(), except that it sends the request and then immediately returns a FutureResponse instead of blocking waiting on a response and returning a pysoa.common.types.JobResponse. Just call result(timeout=None) on the future response to block for an available response. Some of the possible exceptions may be raised when this method is called; others may be raised when the future is used.

Returns

A future from which the job response can later be retrieved

Raises

pysoa.common.transport.errors.PySOATransportError,

call_actions_parallel(service_name: str, actions: Union[Iterable[pysoa.common.types.ActionRequest], Iterable[Dict[str, Any]]], expansions: Dict[str, List[str]] = None, raise_job_errors: bool = True, raise_action_errors: bool = True, catch_transport_errors: bool = False, timeout: Optional[int] = None, switches: Union[List[int], AbstractSet[int], None] = None, correlation_id: Optional[str] = None, context: Union[Dict[str, Any], None] = None, control_extra: Union[Dict[str, Any], None] = None) → Generator[pysoa.common.types.ActionResponse, None, None]

Build and send multiple job requests to one service, each job with one action, to be executed in parallel, and return once all responses have been received.

Returns a list of action responses, one for each action in the same order as provided, or raises an exception if any action response is an error (unless raise_action_errors is passed as False) or if any job response is an error (unless raise_job_errors is passed as False).

This method performs expansions if the Client is configured with an expansion converter.

Parameters

• service_name – The name of the service to call.

• actions – A list of pysoa.common.types.ActionRequest objects and/or dicts that can be converted to ActionRequest objects.

• expansions – A dictionary representing the expansions to perform.

• raise_job_errors – Whether to raise a pysoa.client.errors.CallJobError if the job response contains errors (defaults to True).

• raise_action_errors – Whether to raise a pysoa.client.errors.CallActionError if any action responses contain errors (defaults to True).

• catch_transport_errors – Whether to catch transport errors and return them instead of letting them propagate. By default (False), all raised pysoa.common.transport.errors.PySOATransportError exceptions cause the entire process to terminate, potentially losing responses. If this argument is set to True, those errors are, instead, caught, and they are returned in place of their corresponding responses in the returned list of job responses. You should not do this in most cases, but it is helpful if you really need to get the successful responses even if there are errors getting other responses.

• timeout – If provided, this will override the default transport timeout values to; requests will expire after this number of seconds plus some buffer defined by the transport, and the client will not block waiting for a response for longer than this amount of time.

• switches – A list of switch value integers.

• correlation_id – The request correlation ID.

• context – A dictionary of extra values to include in the context header.

• control_extra – A dictionary of extra values to include in the control header.

Returns

A generator of action responses

Raises

pysoa.common.transport.errors.PySOATransportError, pysoa.client.errors.CallActionError, pysoa.client.errors.CallJobError

call_actions_parallel_future(service_name: str, actions: Union[Iterable[pysoa.common.types.ActionRequest], Iterable[Dict[str, Any]]], expansions: Dict[str, List[str]] = None, raise_job_errors: bool = True, raise_action_errors: bool = True, catch_transport_errors: bool = False, timeout: Optional[int] = None, switches: Union[List[int], AbstractSet[int], None] = None, correlation_id: Optional[str] = None, context: Union[Dict[str, Any], None] = None, control_extra: Union[Dict[str, Any], None] = None) → pysoa.client.client.FutureSOAResponse[Generator[pysoa.common.types.ActionResponse, None, None]]

This method is identical in signature and behavior to call_actions_parallel(), except that it sends the requests and then immediately returns a FutureResponse instead of blocking waiting on responses and returning a generator. Just call result(timeout=None) on the future response to block for an available response (which will be a generator). Some of the possible exceptions may be raised when this method is called; others may be raised when the future is used.

If argument raise_job_errors is supplied and is False, some items in the result list might be lists of job errors instead of individual pysoa.common.types.ActionResponse objects. Be sure to check for that if used in this manner.

If argument catch_transport_errors is supplied and is True, some items in the result list might be instances of Exception instead of individual pysoa.common.types.ActionResponse objects. Be sure to check for that if used in this manner.

Returns

A generator of action responses that blocks waiting on responses once you begin iteration

Raises

pysoa.common.transport.errors.PySOATransportError,

call_jobs_parallel(jobs: Iterable[Dict[str, Any]], expansions: Dict[str, List[str]] = None, raise_job_errors: bool = True, raise_action_errors: bool = True, catch_transport_errors: bool = False, timeout: Optional[int] = None, switches: Union[List[int], AbstractSet[int], None] = None, correlation_id: Optional[str] = None, continue_on_error: bool = False, context: Union[Dict[str, Any], None] = None, control_extra: Union[Dict[str, Any], None] = None) → List[pysoa.common.types.JobResponse]

Build and send multiple job requests to one or more services, each with one or more actions, to be executed in parallel, and return once all responses have been received.

Returns a list of job responses, one for each job in the same order as provided, or raises an exception if any job response is an error (unless raise_job_errors is passed as False) or if any action response is an error (unless raise_action_errors is passed as False).

This method performs expansions if the Client is configured with an expansion converter.

Parameters

• jobs – A list of job request dicts, each containing service_name and actions, where actions is a list of pysoa.common.types.ActionRequest objects and/or dicts that can be converted to ActionRequest objects.

• expansions – A dictionary representing the expansions to perform.

• raise_job_errors – Whether to raise a pysoa.client.errors.CallJobError if the job response contains errors (defaults to True).

• raise_action_errors – Whether to raise a pysoa.client.errors.CallActionError if any action responses contain errors (defaults to True).

• catch_transport_errors – Whether to catch transport errors and return them instead of letting them propagate. By default (False), all raised pysoa.common.transport.errors.PySOATransportError exceptions cause the entire process to terminate, potentially losing responses. If this argument is set to True, those errors are, instead, caught, and they are returned in place of their corresponding responses in the returned list of job responses. You should not do this in most cases, but it is helpful if you really need to get the successful responses even if there are errors getting other responses.

• timeout – If provided, this will override the default transport timeout values to; requests will expire after this number of seconds plus some buffer defined by the transport, and the client will not block waiting for a response for longer than this amount of time.

• switches – A list of switch value integers.

• correlation_id – The request correlation ID.

• continue_on_error – Whether the service should continue executing further actions once one action has returned errors (only applies to multiple actions in a single job).

• context – A dictionary of extra values to include in the context header.

• control_extra – A dictionary of extra values to include in the control header.

Returns

The job response

Raises

pysoa.common.transport.errors.PySOATransportError, pysoa.client.errors.CallActionError, pysoa.client.errors.CallJobError

call_jobs_parallel_future(jobs: Iterable[Dict[str, Any]], expansions: Dict[str, List[str]] = None, raise_job_errors: bool = True, raise_action_errors: bool = True, catch_transport_errors: bool = False, timeout: Optional[int] = None, switches: Union[List[int], AbstractSet[int], None] = None, correlation_id: Optional[str] = None, continue_on_error: bool = False, context: Union[Dict[str, Any], None] = None, control_extra: Union[Dict[str, Any], None] = None) → pysoa.client.client.FutureSOAResponse[List[pysoa.common.types.JobResponse]]

This method is identical in signature and behavior to call_jobs_parallel(), except that it sends the requests and then immediately returns a FutureResponse instead of blocking waiting on all responses and returning a list of pysoa.common.types.JobResponse objects. Just call result(timeout=None) on the future response to block for an available response. Some of the possible exceptions may be raised when this method is called; others may be raised when the future is used.

Returns

A future from which the list of job responses can later be retrieved

Raises

pysoa.common.transport.errors.PySOATransportError

get_all_responses(service_name: str, receive_timeout_in_seconds: Optional[int] = None) → Generator[Tuple[int, pysoa.common.types.JobResponse], None, None]

Receive all available responses from the service as a generator.

Parameters

• service_name – The name of the service from which to receive responses

• receive_timeout_in_seconds – How long to block without receiving a message before raising pysoa.common.transport.errors.MessageReceiveTimeout (defaults to five seconds unless the settings are otherwise).

Returns

A generator that yields a two-tuple of request ID, job response

Raises

pysoa.common.transport.errors.PySOATransportError

handler_class

alias of ServiceHandler

send_request(service_name: str, actions: Union[List[pysoa.common.types.ActionRequest], List[Dict[str, Any]]], switches: Union[List[int], AbstractSet[int], None] = None, correlation_id: Optional[str] = None, continue_on_error: bool = False, context: Union[Dict[str, Any], None] = None, control_extra: Union[Dict[str, Any], None] = None, message_expiry_in_seconds: Optional[int] = None, suppress_response: bool = False) → int

Build and send a JobRequest, and return a request ID.

The context and control_extra arguments may be used to include extra values in the context and control headers, respectively.

Parameters

• service_name – The name of the service from which to receive responses

• actions – A list of ActionRequest objects or dictionaries

• switches – A list of switch value integers

• correlation_id – The request correlation ID

• continue_on_error – Whether to continue executing further actions once one action has returned errors

• context – A dictionary of extra values to include in the context header

• control_extra – A dictionary of extra values to include in the control header

• message_expiry_in_seconds – How soon the message will expire if not received by a server (defaults to sixty seconds unless the settings are otherwise)

• suppress_response – If True, the service will process the request normally but omit the step of sending a response back to the client (use this feature to implement send-and-forget patterns for asynchronous execution)

Returns

The request ID

Raises

pysoa.common.transport.errors.PySOATransportError

settings_class

alias of pysoa.client.settings.ClientSettings

class pysoa.client.client.ServiceHandler(service_name, settings)

Bases: object

Does the low-level work of communicating with an individual service through its configured transport.

__init__(service_name: str, settings: <class 'pysoa.client.settings.ClientSettings'>) → None

Parameters

• service_name – The name of the service which this handler calls

• settings – The client settings object for this service (and only this service)

get_all_responses(receive_timeout_in_seconds: Optional[int] = None) → Generator[Tuple[int, pysoa.common.types.JobResponse], None, None]

Receive all available responses from the transport as a generator.

Parameters

receive_timeout_in_seconds – How long to block without receiving a message before raising pysoa.common.transport.errors.MessageReceiveTimeout (defaults to five seconds unless the settings are otherwise).

Returns

A generator that yields a two-tuple of request ID, job response

Raises

pysoa.common.transport.errors.PySOATransportError, StopIteration

send_request(job_request: <class 'pysoa.common.types.JobRequest'>, message_expiry_in_seconds: Optional[int] = None) → int

Send a JobRequest, and return a request ID.

Parameters

• job_request – The job request object to send

• message_expiry_in_seconds – How soon the message will expire if not received by a server (defaults to sixty seconds unless the settings are otherwise)

Returns

The request ID

Raises

pysoa.common.transport.errors.PySOATransportError

exception pysoa.client.errors.CallActionError(actions=None)

Bases: pysoa.client.errors.PySOAClientError

Raised by Client.call_*** methods when a job response contains one or more action errors. Stores a list of ActionResponse objects and has a string representation cleanly displaying the actions’ errors.

__init__(actions: Union[List[pysoa.common.types.ActionResponse], None] = None) → None

Parameters

actions – The list of all actions that have errors (not actions without errors), available as an actions property on the exception instance.

__str__()

Return str(self).

exception pysoa.client.errors.CallJobError(errors=None)

Bases: pysoa.client.errors.PySOAClientError

Raised by Client.call_*** methods when a job response contains one or more job errors. Stores a list of Error objects and has a string representation cleanly displaying the errors.

__init__(errors: Union[List[pysoa.common.errors.Error], None] = None) → None

Parameters

errors – The list of all errors in this job, available as an errors property on the exception instance.

__str__()

Return str(self).

exception pysoa.client.errors.ImproperlyConfigured

Bases: pysoa.client.errors.PySOAClientError

Raised when this client is improperly configured to call the specified service.

exception pysoa.client.errors.InvalidExpansionKey

Bases: pysoa.client.errors.PySOAClientError

Raised when this client is improperly configured to perform the specified expansion.

exception pysoa.client.errors.PySOAClientError

Bases: pysoa.common.errors.PySOAError

Base exception for all client-side errors other than transport errors.

class pysoa.client.expander.ExpansionSettings(data)

Bases: conformity.settings.Settings

Defines the schema for configuration settings used when expanding objects on responses with the Expansions tool.

Settings Schema Definition

• type_expansions - flexible dict: The definition of all types that may contain identifiers that can be expanded into objects using the type_routes configurations

  keys

  unicode: The name of the type for which the herein defined expansions can be sought, which will be matched with a key from the expansions dict passed to one of Client’s call_*** methods, and which must also match the value of a _type field found on response objects on which extra data will be expanded

  values

  flexible dict: The definition of all possible expansions for this object type

  keys

  unicode: The name of an expansion, which will be matched with a value from the expansions dict passed to one of Client’s call_*** methods corresponding to the type key in that dict

  values

  strict dict: The definition of one specific possible expansion for this object type

  • destination_field - unicode: The name of a not-already-existent field in the base object into which the expansion object will be placed after it is obtained from the route

  • raise_action_errors - boolean: Whether to raise action errors encountered when expanding objects these objects (by default, action errors are suppressed, which differs from the behavior of the Client to raise action errors during normal requests)

  • route - unicode: The route to use to resolve this expansion, which must match a key in the type_routes configuration

  • source_field - unicode: The name of the field in the base object that contains the identifier used for obtaining the expansion object (the identifier will be passed to the request_field in the route when resolving the expansion)

  • type - unicode (nullable): The type of object this expansion yields, which must map back to a type_expansions key in order to support nested/recursive expansions, and may be None if you do not wish to support nested/recursive expansions for this expansion

  Optional keys: raise_action_errors

• type_routes - flexible dict: The definition of all recognized types that can be expanded into and information about how to resolve objects of those types through action calls

  keys

  unicode: The name of the expansion route, to be referenced from the type_expansions configuration

  values

  strict dict: The instructions for resolving this type route

  • action - unicode: The name of the action to call to resolve this route, which must accept a single request field of type List, to which all the identifiers for matching candidate expansions will be passed, and which must return a single response field of type Dictionary, from which all expansion objects will be obtained

  • request_field - unicode: The name of the List identifier field to place in the ActionRequest body when making the request to the named service and action

  • response_field - unicode: The name of the Dictionary field returned in the ActionResponse, from which the expanded objects will be extracted

  • service - unicode: The name of the service to call to resolve this route

Default Values

Keys present in the dict below can be omitted from compliant settings dicts, in which case the values below will apply as the default values.

{}

defaults = {}

schema = {'type_expansions': SchemalessDictionary(key_type=UnicodeString(min_length=None, max_length=None, description="The name of the type for which the herein defined expansions can be sought, which will be matched with a key from the `expansions` dict passed to one of `Client`'s `call_***` methods, and which must also match the value of a `_type` field found on response objects on which extra data will be expanded", allow_blank=True), value_type=SchemalessDictionary(key_type=UnicodeString(min_length=None, max_length=None, description="The name of an expansion, which will be matched with a value from the `expansions` dict passed to one of `Client`'s `call_***` methods corresponding to the type key in that dict", allow_blank=True), value_type=Dictionary(contents={'type': Nullable(field=UnicodeString(min_length=None, max_length=None, description='The type of object this expansion yields, which must map back to a `type_expansions` key in order to support nested/recursive expansions, and may be `None` if you do not wish to support nested/recursive expansions for this expansion', allow_blank=True)), 'route': UnicodeString(min_length=None, max_length=None, description='The route to use to resolve this expansion, which must match a key in the `type_routes` configuration', allow_blank=True), 'source_field': UnicodeString(min_length=None, max_length=None, description='The name of the field in the base object that contains the identifier used for obtaining the expansion object (the identifier will be passed to the `request_field` in the route when resolving the expansion)', allow_blank=True), 'destination_field': UnicodeString(min_length=None, max_length=None, description='The name of a not-already-existent field in the base object into which the expansion object will be placed after it is obtained from the route', allow_blank=True), 'raise_action_errors': Boolean(description='Whether to raise action errors encountered when expanding objects these objects (by default, action errors are suppressed, which differs from the behavior of the `Client` to raise action errors during normal requests)')}, optional_keys=frozenset({'raise_action_errors'}), allow_extra_keys=False, description='The definition of one specific possible expansion for this object type', additional_validator=None), max_length=None, min_length=None, description='The definition of all possible expansions for this object type', additional_validator=None), max_length=None, min_length=None, description='The definition of all types that may contain identifiers that can be expanded into objects using the `type_routes` configurations', additional_validator=None), 'type_routes': SchemalessDictionary(key_type=UnicodeString(min_length=None, max_length=None, description='The name of the expansion route, to be referenced from the `type_expansions` configuration', allow_blank=True), value_type=Dictionary(contents={'service': UnicodeString(min_length=None, max_length=None, description='The name of the service to call to resolve this route', allow_blank=True), 'action': UnicodeString(min_length=None, max_length=None, description='The name of the action to call to resolve this route, which must accept a single request field of type `List`, to which all the identifiers for matching candidate expansions will be passed, and which must return a single response field of type `Dictionary`, from which all expansion objects will be obtained', allow_blank=True), 'request_field': UnicodeString(min_length=None, max_length=None, description='The name of the `List` identifier field to place in the `ActionRequest` body when making the request to the named service and action', allow_blank=True), 'response_field': UnicodeString(min_length=None, max_length=None, description='The name of the `Dictionary` field returned in the `ActionResponse`, from which the expanded objects will be extracted', allow_blank=True)}, optional_keys=frozenset(), allow_extra_keys=False, description='The instructions for resolving this type route', additional_validator=None), max_length=None, min_length=None, description='The definition of all recognized types that can be expanded into and information about how to resolve objects of those types through action calls', additional_validator=None)}

class pysoa.client.middleware.ClientMiddleware

Bases: object

Base middleware class for client middleware. Not required, but provides some helpful stubbed methods and documentation that you should follow for creating your middleware classes. If you extend this class, you may override either one or both of the methods.

Middleware must have two callable attributes, request and response, that, when called with the next level down, return a callable that takes the appropriate arguments and returns the appropriate value.

Class Configuration Schema

strict dict: Most client middleware has no constructor arguments, but subclasses can override this schema

No keys permitted.

request(send_request: Callable[[int, Dict[str, Any], pysoa.common.types.JobRequest, Optional[int]], None]) → Callable[[int, Dict[str, Any], pysoa.common.types.JobRequest, Optional[int]], None]

In sub-classes, used for creating a wrapper around send_request. In this simple implementation, just returns send_request.

Parameters

send_request – A callable that accepts a request ID int, meta dict, pysoa.common.types.JobRequest object, and message expiry int and returns nothing.

Returns

A callable that accepts a request ID int, meta dict, pysoa.common.types.JobRequest object, and message expiry int and returns nothing.

response(get_response: Callable[[Optional[int]], Tuple[Optional[int], Optional[pysoa.common.types.JobResponse]]]) → Callable[[Optional[int]], Tuple[Optional[int], Optional[pysoa.common.types.JobResponse]]]

In sub-classes, used for creating a wrapper around get_response. In this simple implementation, just returns get_response.

Parameters

get_response – A callable that accepts a timeout int and returns tuple of request ID int and pysoa.common.types.JobResponse object.

Returns

A callable that accepts a timeout int and returns tuple of request ID int and pysoa.common.types.JobResponse object.

class pysoa.client.settings.ClientSettings(data)

Bases: pysoa.common.settings.SOASettings

Base settings class for all clients, whose middleware values are restricted to subclasses of ClientMiddleware and whose transport values are restricted to subclasses of BaseClientTransport. Middleware and transport configuration settings schemas will automatically switch based on the configuration settings schema for the path for each.

Settings Schema Definition

• metrics - dictionary with keys path and kwargs whose kwargs schema switches based on the value of path, dynamically based on class imported from path (see the configuration settings schema documentation for the class named at path). Configuration for defining a usage and performance metrics recorder. The imported item at the specified path must be a subclass of pymetrics.recorders.base.MetricsRecorder.

• middleware - list: The list of all ClientMiddleware objects that should be applied to requests made from this client to the associated service

  values

  dictionary with keys path and kwargs whose kwargs schema switches based on the value of path, dynamically based on class imported from path (see the configuration settings schema documentation for the class named at path). The imported item at the specified path must be a subclass of pysoa.client.middleware.ClientMiddleware.

• transport - dictionary with keys path and kwargs whose kwargs schema switches based on the value of path, dynamically based on class imported from path (see the configuration settings schema documentation for the class named at path). The imported item at the specified path must be a subclass of pysoa.common.transport.base.ClientTransport.

Default Values

Keys present in the dict below can be omitted from compliant settings dicts, in which case the values below will apply as the default values.

{
    "metrics": {
        "path": "pymetrics.recorders.noop:NonOperationalMetricsRecorder"
    },
    "middleware": [],
    "transport": {
        "path": "pysoa.common.transport.redis_gateway.client:RedisClientTransport"
    }
}

defaults = {'metrics': {'path': 'pymetrics.recorders.noop:NonOperationalMetricsRecorder'}, 'middleware': [], 'transport': {'path': 'pysoa.common.transport.redis_gateway.client:RedisClientTransport'}}

schema = {'metrics': ClassConfigurationSchema(base_class=<class 'pymetrics.recorders.base.MetricsRecorder'>, default_path=None, description='Configuration for defining a usage and performance metrics recorder.', eager_default_validation=True, add_class_object_to_dict=True), 'middleware': List(contents=ClassConfigurationSchema(base_class=<class 'pysoa.client.middleware.ClientMiddleware'>, default_path=None, description=None, eager_default_validation=True, add_class_object_to_dict=True), max_length=None, min_length=None, description='The list of all `ClientMiddleware` objects that should be applied to requests made from this client to the associated service', additional_validator=None), 'transport': ClassConfigurationSchema(base_class=<class 'pysoa.common.transport.base.ClientTransport'>, default_path=None, description=None, eager_default_validation=True, add_class_object_to_dict=True)}

class pysoa.common.errors.Error(code: str, message: str, field: Optional[str] = None, traceback: Optional[str] = None, variables: Union[Dict[str, Any], None] = None, denied_permissions: Union[List[str], None] = None, is_caller_error: bool = False)

Bases: object

Represents an error that occurred, in the format transmitted over the transport between client and service.

Parameters

• code – The machine-readable error code.

• message – The human-readable error message.

• field – If the error is the result of validation of a job attribute or action request field, this contains the name of that attribute or field. It will be None otherwise.

• traceback – If the error is the result of an exception with valuable traceback information, this contains that traceback. It will be None otherwise.

• variables – If there are any variables pertinent to this error, this dictionary will contain the names and values of those variables. It will be None otherwise. New in version 0.9.0.

• denied_permissions – If this error is the result of insufficient privileges to perform the requested operation, this attribute will be a list containing the necessary permissions that were denied, if the service supports reporting this information. It will be None otherwise. New in version 0.44.0.

• is_caller_error – Indicates whether this error is the result of the caller doing something wrong (True), such as an invalid job request or an action request body that fails to validate, or the result of a server or service problem or bug (False). New in version 0.70.0.

__delattr__(name)

Attached to frozen classes as __delattr__.

__init__(code: str, message: str, field: Optional[str] = None, traceback: Optional[str] = None, variables: Optional[Dict[str, Any]] = None, denied_permissions: Optional[List[str]] = None, is_caller_error: bool = False)

Initialize self. See help(type(self)) for accurate signature.

__setattr__(name, value)

Attached to frozen classes as __setattr__.

exception pysoa.common.errors.PySOAError

Bases: Exception

Base exception for all PySOA errors.

class pysoa.common.settings.SOASettings(data)

Bases: conformity.settings.Settings

Settings shared between client and server.

Settings Schema Definition

• metrics - dictionary with keys path and kwargs whose kwargs schema switches based on the value of path, dynamically based on class imported from path (see the configuration settings schema documentation for the class named at path). Configuration for defining a usage and performance metrics recorder. The imported item at the specified path must be a subclass of pymetrics.recorders.base.MetricsRecorder.

• middleware - list: The list of all middleware objects that should be applied to this server or client

  values

  dictionary with keys path and kwargs whose kwargs schema switches based on the value of path, dynamically based on class imported from path (see the configuration settings schema documentation for the class named at path). The imported item at the specified path must be a subclass of builtins.object.

• transport - dictionary with keys path and kwargs whose kwargs schema switches based on the value of path, dynamically based on class imported from path (see the configuration settings schema documentation for the class named at path). The imported item at the specified path must be a subclass of builtins.object.

Default Values

Keys present in the dict below can be omitted from compliant settings dicts, in which case the values below will apply as the default values.

{
    "metrics": {
        "path": "pymetrics.recorders.noop:NonOperationalMetricsRecorder"
    },
    "middleware": []
}

defaults = {'metrics': {'path': 'pymetrics.recorders.noop:NonOperationalMetricsRecorder'}, 'middleware': []}

schema = {'metrics': ClassConfigurationSchema(base_class=<class 'pymetrics.recorders.base.MetricsRecorder'>, default_path=None, description='Configuration for defining a usage and performance metrics recorder.', eager_default_validation=True, add_class_object_to_dict=True), 'middleware': List(contents=ClassConfigurationSchema(base_class=<class 'object'>, default_path=None, description=None, eager_default_validation=True, add_class_object_to_dict=True), max_length=None, min_length=None, description='The list of all middleware objects that should be applied to this server or client', additional_validator=None), 'transport': ClassConfigurationSchema(base_class=<class 'object'>, default_path=None, description=None, eager_default_validation=True, add_class_object_to_dict=True)}

class pysoa.common.types.ActionRequest(action: str, body: Dict[str, Any] = NOTHING)

Bases: object

A request that the server execute a single action.

Parameters

• action – The name of the action to execute.

• body – The request body input.

__init__(action: str, body: Dict[str, Any] = NOTHING)

Initialize self. See help(type(self)) for accurate signature.

class pysoa.common.types.ActionResponse(action: str, errors: List[pysoa.common.errors.Error] = NOTHING, body: Dict[str, Any] = NOTHING)

Bases: object

A response generated by a single action on the server.

Parameters

• action – The name of the action that was executed.

• body – The response body output.

• errors – A list of any action Error or errors that occurred, or an empty list if no errors occurred.

__init__(action: str, errors: List[pysoa.common.errors.Error] = NOTHING, body: Dict[str, Any] = NOTHING)

Initialize self. See help(type(self)) for accurate signature.

pysoa.common.types.Body = typing.Dict[str, typing.Any]

A type used for annotating attributes and arguments that represent action request and response bodies.

pysoa.common.types.Context = typing.Dict[str, typing.Any]

A type used for annotating attributes and arguments that represent job request and response context headers.

pysoa.common.types.Control = typing.Dict[str, typing.Any]

A type used for annotating attributes and arguments that represent job request control headers.

class pysoa.common.types.JobRequest(control: Dict[str, Any] = NOTHING, context: Dict[str, Any] = NOTHING, actions: List[pysoa.common.types.ActionRequest] = NOTHING)

Bases: object

A request that the server execute a job.

A job consists of one or more actions and context and control headers. Each action is an ActionRequest, while the context and control headers are dictionaries.

Parameters

• control – The control header dictionary.

• context – The context header dictionary.

• actions – The list of ActionRequest objects to be executed. This list must contain at least one request.

__init__(control: Dict[str, Any] = NOTHING, context: Dict[str, Any] = NOTHING, actions: List[pysoa.common.types.ActionRequest] = NOTHING)

Initialize self. See help(type(self)) for accurate signature.

class pysoa.common.types.JobResponse(errors: List[pysoa.common.errors.Error] = NOTHING, context: Dict[str, Any] = NOTHING, actions: List[pysoa.common.types.ActionResponse] = NOTHING)

Bases: object

A response generated by a server job.

Contains the result or error generated by each action in the job.

Parameters

• context – The context header dictionary.

• actions – The list of ActionResponse objects that were executed. This list may be shorter than the number of actions in the JobRequest (even empty) if the errors attribute contains one or more errors.

• errors – A list of any job Error or errors that occurred, or an empty list of no errors occurred.

__init__(errors: List[pysoa.common.errors.Error] = NOTHING, context: Dict[str, Any] = NOTHING, actions: List[pysoa.common.types.ActionResponse] = NOTHING)

Initialize self. See help(type(self)) for accurate signature.

class pysoa.common.types.UnicodeKeysDict

Bases: dict

A special dictionary factory used to ensure that Attrs object-to-dict dictionaries contain keys that are only unicode strings instead of byte strings. This factory will become unnecessary and be removed in PySOA 2.0.0 when all Python 2 support is removed.

__setitem__(key: str, value: Any) → None

Set self[key] to value.

setdefault(key: str, default: Optional[~_VT] = None) → ~_VT

Insert key with a value of default if key is not in the dictionary.

Return the value for key if key is in the dictionary, else default.

class pysoa.common.serializer.base.Serializer

Bases: object

The mime type that this serializer supports.

abstract blob_to_dict(blob: bytes) → Dict

Take a serialized message in the form of bytes (string) and return a dict.

Parameters

blob – The blob to deserialize into a message

Returns

The deserialized message.

abstract dict_to_blob(message_dict: Dict) → bytes

Take a message in the form of a dict and return a serialized message in the form of bytes (string).

Parameters

message_dict – The message to serialize into a blob.

Returns

The serialized blob.

mime_type = None

classmethod resolve_serializer(mime_type: str) → <class 'pysoa.common.serializer.base.Serializer'>

Given the requested mime type, return an initialized Serializer that understands that mime type.

Parameters

mime_type – The mime type for which to get a compatible Serializer

Returns

A compatible Serializer.

Raises

ValueError if there is no Serializer that understands this mime type.

exception pysoa.common.serializer.errors.InvalidField

Bases: pysoa.common.serializer.errors.SerializationError

Raised when a field in a message is not serializable.

exception pysoa.common.serializer.errors.InvalidMessage

Bases: pysoa.common.serializer.errors.SerializationError

Raised when a serialized message is incapable of being deserialized.

exception pysoa.common.serializer.errors.SerializationError

Bases: pysoa.common.errors.PySOAError

Base exceptions for all exceptions related to serialization and deserialization.

Transports are the interface between the Client or Server and the transport backend.

Two base classes are provided, with methods split between Client and Server side. In many cases, Transport implementations will inherit from both ClientTransport and ServerTransport and implement both sets of methods, in order to consolidate shared backend code into a single class.

All Transport methods either accept or return a metadata argument. This should be a dict that includes any information that is necessary for processing the message, but is not business logic. For example, if your implementation has multiple serializer types, the metadata may include a mime type to tell the endpoint receiving the message which type of serializer to use.

class pysoa.common.transport.base.ClientTransport(service_name, metrics=<pymetrics.recorders.noop.NonOperationalMetricsRecorder object>)

Bases: pysoa.common.transport.base.Transport

The base client transport defining the interface for transacting PySOA payloads on the client side.

abstract receive_response_message(receive_timeout_in_seconds: Optional[int] = None) → <class 'pysoa.common.transport.base.ReceivedMessage'>

Receive a response message from the backend and return a 3-tuple of (request_id, meta dict, message dict).

Parameters

receive_timeout_in_seconds – How long to block waiting for a response to become available (implementations should provide a sane default or setting for default)

Returns

A named tuple ReceivedMessage of the request ID, meta dict, and message dict, in that order

Raise

ConnectionError, MessageReceiveError, MessageReceiveTimeout

abstract send_request_message(request_id: int, meta: Dict[str, Any], body: Dict[str, Any], message_expiry_in_seconds: Optional[int] = None) → None

Send a request message.

Parameters

• request_id – The request ID

• meta – Meta information about the message

• body – The message body

• message_expiry_in_seconds – How soon the message should expire if not retrieved by a server (implementations should provide a sane default or setting for default)

Raise

ConnectionError, MessageSendError, MessageSendTimeout, MessageTooLarge

pysoa.common.transport.base.get_hex_thread_id() → str

class pysoa.common.transport.base.ReceivedMessage(request_id, meta, body)

Bases: tuple

The representation of a message received through a transport.

__getnewargs__()

Return self as a plain tuple. Used by copy and pickle.

static __new__(_cls, request_id, meta, body)

Create new instance of ReceivedMessage(request_id, meta, body)

property body

Alias for field number 2

property meta

Alias for field number 1

property request_id

Alias for field number 0

class pysoa.common.transport.base.ServerTransport(service_name, metrics=<pymetrics.recorders.noop.NonOperationalMetricsRecorder object>, instance_index=1)

Bases: pysoa.common.transport.base.Transport

The base server transport defining the interface for transacting PySOA payloads on the server side.

Parameters

• service_name – The name of the service for which this transport will receive requests and send responses

• metrics – The optional metrics recorder

• instance_index – The 1-based index of this process among multiple forks

__init__(service_name: str, metrics: <class 'pymetrics.recorders.base.MetricsRecorder'> = <pymetrics.recorders.noop.NonOperationalMetricsRecorder object>, instance_index: int = 1) → None

Parameters

• service_name – The name of the service to which this transport will send requests (and from which it will receive responses)

• metrics – The optional metrics recorder

abstract receive_request_message() → <class 'pysoa.common.transport.base.ReceivedMessage'>

Receive a request message from the backend and return a 3-tuple of (request_id, meta dict, message dict). The metadata may include client reply-to information that should be passed back to send_response_message.

Returns

A named tuple ReceivedMessage of the request ID, meta dict, and message dict, in that order

Raise

ConnectionError, MessageReceiveError, MessageReceiveTimeout

abstract send_response_message(request_id: int, meta: Dict[str, Any], body: Dict[str, Any]) → None

Send a response message. The meta dict returned by receive_request_message should be passed verbatim as the second argument.

Parameters

• request_id – The request ID

• meta – Meta information about the message

• body – The message body

Raise

ConnectionError, MessageSendError, MessageSendTimeout, MessageTooLarge

class pysoa.common.transport.base.Transport(service_name, metrics=<pymetrics.recorders.noop.NonOperationalMetricsRecorder object>)

Bases: object

A base transport from which all client and server transports inherit, establishing base metrics and service name attributes.

__init__(service_name: str, metrics: <class 'pymetrics.recorders.base.MetricsRecorder'> = <pymetrics.recorders.noop.NonOperationalMetricsRecorder object>) → None

Parameters

• service_name – The name of the service to which this transport will send requests (and from which it will receive responses)

• metrics – The optional metrics recorder

exception pysoa.common.transport.errors.ConnectionError

Bases: pysoa.common.transport.errors.TransientPySOATransportError

Raised when the transport cannot obtain the necessary connection to send or receive a message.

exception pysoa.common.transport.errors.InvalidMessageError

Bases: pysoa.common.transport.errors.PySOATransportError

Raised when the transport cannot send a message because there is some problem with its contents or the way it is structured. This is unrelated to serialization and indicates a client- or server-side (wherever it was raised) programming error that must be resolved.

exception pysoa.common.transport.errors.MessageReceiveError

Bases: pysoa.common.transport.errors.TransientPySOATransportError

Raised when an error occurs while the transport is attempting to receive a message. The meaning of such an error can vary wildly depending on the configured transport. See the transport documentation for more information.

exception pysoa.common.transport.errors.MessageReceiveTimeout

Bases: pysoa.common.transport.errors.TransientPySOATransportError

Raised when the transport reaches the timeout waiting to receive a message. On the server side, this is used for application control flow: When the server does not receive a message within the specified time, it performs idle cleanup operations and then asks the transport for a message again, in a loop, indefinitely. On the client side, this means the server did not respond within the specified timeout, and can be the result of several things:

• The client timeout was set too low for the given action or actions called.

• The server action code is poorly performant and is taking too long to respond.

• The server is overloaded and is receiving requests faster than it can process them.

• The server is currently down.

exception pysoa.common.transport.errors.MessageSendError

Bases: pysoa.common.transport.errors.TransientPySOATransportError

Raised when an error occurs while the transport is attempting to send a message. The meaning of such an error can vary wildly depending on the configured transport. See the transport documentation for more information.

exception pysoa.common.transport.errors.MessageSendTimeout

Bases: pysoa.common.transport.errors.TransientPySOATransportError

Raised when the transport encounters a timeout while attempting to send a message. The meaning of such an error can vary wildly depending on the configured transport. See the transport documentation for more information.

exception pysoa.common.transport.errors.MessageTooLarge(message_size_in_bytes, *args)

Bases: pysoa.common.transport.errors.PySOATransportError

Raised when a message is too large to be sent by the configured transport. This indicates a client- or server-side (wherever it was raised) programming error that must be resolved.

__init__(message_size_in_bytes: int, *args: Any) → None

Initialize self. See help(type(self)) for accurate signature.

exception pysoa.common.transport.errors.PySOATransportError

Bases: pysoa.common.errors.PySOAError

Base exception for all transport-related PySOA errors.

exception pysoa.common.transport.errors.TransientPySOATransportError

Bases: pysoa.common.transport.errors.PySOATransportError

Base exception for transport errors that are typically transient (a failure to send or receive an error) and cannot (usually) be resolved my changes to the client- or server-side code.

class pysoa.common.transport.local.LocalClientTransport(service_name, metrics, server_class, server_settings)

Bases: pysoa.common.transport.base.ClientTransport, pysoa.common.transport.base.ServerTransport

A transport that incorporates a server for running a service and client in a single thread.

Class Configuration Schema

strict dict: The constructor kwargs for the local client transport.

• server_class - any of the types bulleted below: The path to the Server class to use locally (as a library), or a reference to the Server-extending class/type itself.

  • a unicode string importable Python path in the format “foo.bar.MyClass”, “foo.bar:YourClass.CONSTANT”, etc. The importable Python path to the Server-extending class. The imported item at the specified path must match the following schema:

    schema

    a Python type that is a subclass of the following class or classes: pysoa.server.server.Server.

  • a Python type that is a subclass of the following class or classes: pysoa.server.server.Server. A reference to the Server-extending class

• server_settings - any of the types bulleted below: The settings to use when instantiating the server_class.

  • a unicode string importable Python path in the format “foo.bar.MyClass”, “foo.bar:YourClass.CONSTANT”, etc. The importable Python path to the settings dict, in the format “module.name:VARIABLE”. The imported item at the specified path must match the following schema:

    schema

    flexible dict: A dictionary of settings for the server (which will further validate them).

    keys

    unicode: (no description)

    values

    anything: (no description)

  • flexible dict: A dictionary of settings for the server (which will further validate them).

    keys

    unicode: (no description)

    values

    anything: (no description)

__init__(service_name: str, metrics: <class 'pymetrics.recorders.base.MetricsRecorder'>, server_class: Union[str, Type[pysoa.server.server.Server]], server_settings: Union[str, Dict[str, Any]]) → None

Parameters

• service_name – The service name

• metrics – The metrics recorder

• server_class – The server class for which this transport will serve as a client

• server_settings – The server settings that will be passed to the server class on instantiation

receive_request_message() → <class 'pysoa.common.transport.base.ReceivedMessage'>

Gives the server the current request (we are actually inside the stack of send_request_message so we know this is OK).

receive_response_message(_: Optional[int] = None) → <class 'pysoa.common.transport.base.ReceivedMessage'>

Receives a message from the deque. receive_timeout_in_seconds is not supported. Receive does not time out, because by the time the thread calls this method, a response is already available in the deque, or something happened and a response will never be available. This method does not wait and returns immediately.

send_request_message(request_id: int, meta: Dict[str, Any], body: Dict[str, Any], _: Optional[int] = None) → None

Receives a request from the client and handles and dispatches in in-thread. message_expiry_in_seconds is not supported. Messages do not expire, as the server handles the request immediately in the same thread before this method returns. This method blocks until the server has completed handling the request.

send_response_message(request_id: int, meta: Dict[str, Any], body: Dict[str, Any]) → None

Add the response to the deque.

class pysoa.common.transport.local.LocalClientTransportSchema(contents=None, optional_keys=frozenset({}), allow_extra_keys=None, description=None, additional_validator=None)

Bases: conformity.fields.structures.Dictionary

contents = {'server_class': Any(), 'server_settings': Any()}

description = 'The constructor kwargs for the local client transport.'

class pysoa.common.transport.local.LocalServerTransport(service_name, metrics=<pymetrics.recorders.noop.NonOperationalMetricsRecorder object>, instance_index=1)

Bases: pysoa.common.transport.base.ServerTransport

Empty class that we use as an import stub for local transport before we swap in the Client transport instance to do double duty.

Class Configuration Schema

strict dict: The local server transport takes no constructor kwargs.

No keys permitted.

receive_request_message()

Does nothing, because this will never be called (the same-named method on the LocalClientTransport is called, instead).

send_response_message(request_id, meta, body)

Does nothing, because this will never be called (the same-named method on the LocalClientTransport is called, instead).

class pysoa.common.transport.local.LocalServerTransportSchema(contents=None, optional_keys=frozenset({}), allow_extra_keys=None, description=None, additional_validator=None)

Bases: conformity.fields.structures.Dictionary

contents = {}

description = 'The local server transport takes no constructor kwargs.'

class pysoa.common.transport.redis_gateway.client.RedisClientTransport(service_name, metrics, **kwargs)

Bases: pysoa.common.transport.base.ClientTransport

Class Configuration Schema

strict dict: The constructor kwargs for the Redis client transport.

• backend_layer_kwargs - strict dict: The arguments passed to the Redis connection manager

  • connection_kwargs - flexible dict: The arguments used when creating all Redis connections (see Redis-Py docs)

    keys

    hashable: (no description)

    values

    anything: (no description)

  • hosts - list: The list of Redis hosts, where each is a tuple of ("address", port) or the simple string address.

    values

    any of the types bulleted below: (no description)

    • tuple: (no description) (additional information: {'contents': [{'type': 'unicode'}, {'type': 'integer'}]})

    • unicode: (no description)

  • redis_db - integer: The Redis database, a shortcut for putting this in connection_kwargs.

  • redis_port - integer: The port number, a shortcut for putting this on all hosts

  • sentinel_failover_retries - integer: How many times to retry (with a delay) getting a connection from the Sentinel when a master cannot be found (cluster is in the middle of a failover); should only be used for Sentinel backend type

  • sentinel_kwargs - flexible dict: The arguments used when creating all Sentinel connections (see Redis-Py docs); should only be used for Sentinel backend type; similar to connection_kwargs, but you may need to specify both (one for Sentinel connections, one for Redis connections)

    keys

    hashable: (no description)

    values

    anything: (no description)

  • sentinel_services - list: A list of Sentinel services (will be discovered by default); should only be used for Sentinel backend type

    values

    unicode: (no description)

  Optional keys: connection_kwargs, hosts, redis_db, redis_port, sentinel_failover_retries, sentinel_kwargs, sentinel_services

• backend_type - constant: Which backend (standard or sentinel) should be used for this Redis transport (additional information: {'values': ['redis.sentinel', 'redis.standard']})

• default_serializer_config - dictionary with keys path and kwargs whose kwargs schema switches based on the value of path, dynamically based on class imported from path (see the configuration settings schema documentation for the class named at path). The configuration for the serializer this transport should use. The imported item at the specified path must be a subclass of pysoa.common.serializer.base.Serializer.

• log_messages_larger_than_bytes - integer: By default, messages larger than 100KB that do not trigger errors (see maximum_message_size_in_bytes) will be logged with level WARNING to a logger named pysoa.transport.oversized_message. To disable this behavior, set this setting to 0. Or, you can set it to some other number to change the threshold that triggers logging.

• maximum_message_size_in_bytes - integer: The maximum message size, in bytes, that is permitted to be transmitted over this transport (defaults to 100KB on the client and 250KB on the server)

• message_expiry_in_seconds - integer: How long after a message is sent that it is considered expired, dropped from queue

• protocol_version - any of the types bulleted below: The default protocol version between clients and servers was Version 1 prior to PySOA 0.67.0, Version 2 as of 0.67.0, and will be Version 3 as of 1.0.0. The server can only detect what protocol the client is speaking and respond with the same protocol. However, the client cannot pre-determine what protocol the server is speaking. So, if you need to differ from the default (currently Version 2), use this setting to tell the client which protocol to speak.

  • integer: (no description)

  • a Python object that is an instance of the following class or classes: pysoa.common.transport.redis_gateway.constants.ProtocolVersion.

• queue_capacity - integer: The capacity of the message queue to which this transport will send messages

• queue_full_retries - integer: How many times to retry sending a message to a full queue before giving up

• receive_timeout_in_seconds - integer: How long to block waiting on a message to be received

Optional keys: backend_layer_kwargs, default_serializer_config, log_messages_larger_than_bytes, maximum_message_size_in_bytes, message_expiry_in_seconds, protocol_version, queue_capacity, queue_full_retries, receive_timeout_in_seconds

__init__(service_name: str, metrics: <class 'pymetrics.recorders.base.MetricsRecorder'>, **kwargs: Any) → None

In addition to the two named positional arguments, this constructor expects keyword arguments abiding by the Redis transport settings schema.

Parameters

• service_name – The name of the service to which this transport will send requests (and from which it will receive responses)

• metrics – The optional metrics recorder

receive_response_message(receive_timeout_in_seconds: Optional[int] = None) → <class 'pysoa.common.transport.base.ReceivedMessage'>

Receive a response message from the backend and return a 3-tuple of (request_id, meta dict, message dict).

Parameters

receive_timeout_in_seconds – How long to block waiting for a response to become available (implementations should provide a sane default or setting for default)

Returns

A named tuple ReceivedMessage of the request ID, meta dict, and message dict, in that order

Raise

ConnectionError, MessageReceiveError, MessageReceiveTimeout

property requests_outstanding

Indicates the number of requests currently outstanding, which still need to be received. If this value is less than 1, calling receive_response_message will result in a return value of (None, None, None) instead of raising a MessageReceiveTimeout.

send_request_message(request_id: int, meta: Dict[str, Any], body: Dict[str, Any], message_expiry_in_seconds: Optional[int] = None) → None

Send a request message.

Parameters

• request_id – The request ID

• meta – Meta information about the message

• body – The message body

• message_expiry_in_seconds – How soon the message should expire if not retrieved by a server (implementations should provide a sane default or setting for default)

Raise

ConnectionError, MessageSendError, MessageSendTimeout, MessageTooLarge

class pysoa.common.transport.redis_gateway.server.RedisServerTransport(service_name, metrics, instance_index, **kwargs)

Bases: pysoa.common.transport.base.ServerTransport

Class Configuration Schema

strict dict: The constructor kwargs for the Redis server transport.

• backend_layer_kwargs - strict dict: The arguments passed to the Redis connection manager

  • connection_kwargs - flexible dict: The arguments used when creating all Redis connections (see Redis-Py docs)

    keys

    hashable: (no description)

    values

    anything: (no description)

  • hosts - list: The list of Redis hosts, where each is a tuple of ("address", port) or the simple string address.

    values

    any of the types bulleted below: (no description)

    • tuple: (no description) (additional information: {'contents': [{'type': 'unicode'}, {'type': 'integer'}]})

    • unicode: (no description)

  • redis_db - integer: The Redis database, a shortcut for putting this in connection_kwargs.

  • redis_port - integer: The port number, a shortcut for putting this on all hosts

  • sentinel_failover_retries - integer: How many times to retry (with a delay) getting a connection from the Sentinel when a master cannot be found (cluster is in the middle of a failover); should only be used for Sentinel backend type

  • sentinel_kwargs - flexible dict: The arguments used when creating all Sentinel connections (see Redis-Py docs); should only be used for Sentinel backend type; similar to connection_kwargs, but you may need to specify both (one for Sentinel connections, one for Redis connections)

    keys

    hashable: (no description)

    values

    anything: (no description)

  • sentinel_services - list: A list of Sentinel services (will be discovered by default); should only be used for Sentinel backend type

    values

    unicode: (no description)

  Optional keys: connection_kwargs, hosts, redis_db, redis_port, sentinel_failover_retries, sentinel_kwargs, sentinel_services

• backend_type - constant: Which backend (standard or sentinel) should be used for this Redis transport (additional information: {'values': ['redis.sentinel', 'redis.standard']})

• chunk_messages_larger_than_bytes - integer: If set, responses larger than this setting will be chunked and sent back to the client in pieces, to prevent blocking single-threaded Redis for long periods of time to handle large responses. When set, this value must be greater than or equal to 102400, and maximum_message_size_in_bytes must also be set and must be at least 5 times greater than this value (because maximum_message_size_in_bytes is still enforced).

• default_serializer_config - dictionary with keys path and kwargs whose kwargs schema switches based on the value of path, dynamically based on class imported from path (see the configuration settings schema documentation for the class named at path). The configuration for the serializer this transport should use. The imported item at the specified path must be a subclass of pysoa.common.serializer.base.Serializer.

• log_messages_larger_than_bytes - integer: By default, messages larger than 100KB that do not trigger errors (see maximum_message_size_in_bytes) will be logged with level WARNING to a logger named pysoa.transport.oversized_message. To disable this behavior, set this setting to 0. Or, you can set it to some other number to change the threshold that triggers logging.

• maximum_message_size_in_bytes - integer: The maximum message size, in bytes, that is permitted to be transmitted over this transport (defaults to 100KB on the client and 250KB on the server)

• message_expiry_in_seconds - integer: How long after a message is sent that it is considered expired, dropped from queue

• queue_capacity - integer: The capacity of the message queue to which this transport will send messages

• queue_full_retries - integer: How many times to retry sending a message to a full queue before giving up

• receive_timeout_in_seconds - integer: How long to block waiting on a message to be received

Optional keys: backend_layer_kwargs, chunk_messages_larger_than_bytes, default_serializer_config, log_messages_larger_than_bytes, maximum_message_size_in_bytes, message_expiry_in_seconds, queue_capacity, queue_full_retries, receive_timeout_in_seconds

__init__(service_name: str, metrics: <class 'pymetrics.recorders.base.MetricsRecorder'>, instance_index: int, **kwargs: Any) → None

In addition to the named positional arguments, this constructor expects keyword arguments abiding by the Redis transport settings schema.

Parameters

• service_name – The name of the service for which this transport will receive requests and send responses

• metrics – The optional metrics recorder

• instance_index – The 1-based index of this process among multiple forks

receive_request_message() → <class 'pysoa.common.transport.base.ReceivedMessage'>

Receive a request message from the backend and return a 3-tuple of (request_id, meta dict, message dict). The metadata may include client reply-to information that should be passed back to send_response_message.

Returns

A named tuple ReceivedMessage of the request ID, meta dict, and message dict, in that order

Raise

ConnectionError, MessageReceiveError, MessageReceiveTimeout

send_response_message(request_id: int, meta: Dict[str, Any], body: Dict[str, Any]) → None

Send a response message. The meta dict returned by receive_request_message should be passed verbatim as the second argument.

Parameters

• request_id – The request ID

• meta – Meta information about the message

• body – The message body

Raise

ConnectionError, MessageSendError, MessageSendTimeout, MessageTooLarge

class pysoa.common.logging.PySOALogContextFilter

Bases: logging.Filter

__init__() → None

Initialize a filter.

Initialize with the name of the logger which, together with its children, will have its events allowed through the filter. If no name is specified, allow every event.

classmethod clear_logging_action_name() → None

classmethod clear_logging_request_context() → None

filter(record: <class 'logging.LogRecord'>) → bool

Determine if the specified record is to be logged.

Is the specified record to be logged? Returns 0 for no, nonzero for yes. If deemed appropriate, the record may be modified in-place.

classmethod get_logging_action_name() → Optional[str]

classmethod get_logging_request_context() → Union[Dict[str, Any], None]

classmethod set_logging_action_name(action_name: str) → None

classmethod set_logging_request_context(**context: Any) → None

classmethod set_service_name(service_name: str) → None

class pysoa.common.logging.RecursivelyCensoredDictWrapper(wrapped_dict)

Bases: object

CENSORED_STRING = '**********'

CENSOR_TYPES = (<class 'str'>, <class 'int'>)

SENSITIVE_FIELDS = frozenset({'account-number', 'account-numbers', 'accountNumber', 'accountNumbers', 'account_number', 'account_numbers', 'auth', 'auth-token', 'authToken', 'auth_token', 'authentication', 'authentication-token', 'authenticationToken', 'authentication_token', 'authorization', 'authorization-token', 'authorizationToken', 'authorization_token', 'bank-account', 'bank-account-number', 'bank-account-numbers', 'bank-accounts', 'bankAccount', 'bankAccountNumber', 'bankAccountNumbers', 'bankAccounts', 'bank_account', 'bank_account_number', 'bank_account_numbers', 'bank_accounts', 'card-id', 'card-identification', 'card-identification-code', 'card-identification-codes', 'card-identification-number', 'card-identification-numbers', 'card-identification-value', 'card-identification-values', 'card-identifications', 'card-ids', 'card-number', 'card-numbers', 'card-securities', 'card-security', 'card-security-code', 'card-security-codes', 'card-security-number', 'card-security-numbers', 'card-security-value', 'card-security-values', 'card-validation', 'card-validation-code', 'card-validation-codes', 'card-validation-number', 'card-validation-numbers', 'card-validation-value', 'card-validation-values', 'card-validations', 'card-verification', 'card-verification-code', 'card-verification-codes', 'card-verification-number', 'card-verification-numbers', 'card-verification-value', 'card-verification-values', 'card-verifications', 'cardId', 'cardIdentification', 'cardIdentificationCode', 'cardIdentificationCodes', 'cardIdentificationNumber', 'cardIdentificationNumbers', 'cardIdentificationValue', 'cardIdentificationValues', 'cardIdentifications', 'cardIds', 'cardNumber', 'cardNumbers', 'cardSecurities', 'cardSecurity', 'cardSecurityCode', 'cardSecurityCodes', 'cardSecurityNumber', 'cardSecurityNumbers', 'cardSecurityValue', 'cardSecurityValues', 'cardValidation', 'cardValidationCode', 'cardValidationCodes', 'cardValidationNumber', 'cardValidationNumbers', 'cardValidationValue', 'cardValidationValues', 'cardValidations', 'cardVerification', 'cardVerificationCode', 'cardVerificationCodes', 'cardVerificationNumber', 'cardVerificationNumbers', 'cardVerificationValue', 'cardVerificationValues', 'cardVerifications', 'card_id', 'card_identification', 'card_identification_code', 'card_identification_codes', 'card_identification_number', 'card_identification_numbers', 'card_identification_value', 'card_identification_values', 'card_identifications', 'card_ids', 'card_number', 'card_numbers', 'card_securities', 'card_security', 'card_security_code', 'card_security_codes', 'card_security_number', 'card_security_numbers', 'card_security_value', 'card_security_values', 'card_validation', 'card_validation_code', 'card_validation_codes', 'card_validation_number', 'card_validation_numbers', 'card_validation_value', 'card_validation_values', 'card_validations', 'card_verification', 'card_verification_code', 'card_verification_codes', 'card_verification_number', 'card_verification_numbers', 'card_verification_value', 'card_verification_values', 'card_verifications', 'cc-number', 'cc-numbers', 'ccNumber', 'ccNumbers', 'cc_number', 'cc_numbers', 'ccn', 'ccns', 'cid', 'cids', 'credit-card', 'credit-card-number', 'credit-card-numbers', 'credit-cards', 'creditCard', 'creditCardNumber', 'creditCardNumbers', 'creditCards', 'credit_card', 'credit_card_number', 'credit_card_numbers', 'credit_cards', 'csc', 'cscs', 'csn', 'csns', 'cvc', 'cvc2', 'cvc2s', 'cvcs', 'cvd', 'cvds', 'cve', 'cves', 'cvn2', 'cvn2s', 'cvv', 'cvv2', 'cvv2s', 'cvvs', 'icvv', 'icvvs', 'pass', 'pass-phrase', 'pass-phrases', 'passPhrase', 'passPhrases', 'pass_phrase', 'pass_phrases', 'passphrase', 'passphrases', 'passwd', 'passwds', 'password', 'passwords', 'personal-id-number', 'personal-id-numbers', 'personal-identification-number', 'personal-identification-numbers', 'personalIdNumber', 'personalIdNumbers', 'personalIdentificationNumber', 'personalIdentificationNumbers', 'personal_id_number', 'personal_id_numbers', 'personal_identification_number', 'personal_identification_numbers', 'pin', 'pin-code', 'pin-codes', 'pin-number', 'pin-numbers', 'pinCode', 'pinCodes', 'pinNumber', 'pinNumbers', 'pin_code', 'pin_codes', 'pin_number', 'pin_numbers', 'pins', 'private', 'private-key', 'private-keys', 'privateKey', 'privateKeys', 'private_key', 'private_keys', 'privates', 'secret', 'secret-key', 'secret-keys', 'secretKey', 'secretKeys', 'secret_key', 'secret_keys', 'secrets', 'security-code', 'security-codes', 'securityCode', 'securityCodes', 'security_code', 'security_codes', 'token'})

__bytes__() → bytes

__init__(wrapped_dict: Mapping[str, Any]) → None

Wraps a dict to censor its contents. The first time repr is called, it copies the dict, recursively censors sensitive fields, caches the result, and returns the censored dict repr-ed. All future calls use the cache.

Parameters

wrapped_dict – The dict that should be censored

__str__() → str

Return str(self).

__unicode__() → str

class pysoa.common.logging.SyslogHandler(address=('localhost', 514), facility=1, socket_type=None, overflow=0)

Bases: logging.handlers.SysLogHandler

A more advanced Syslog logging handler that attempts to understand the MTU of the underlying connection and then tailor Syslog packets to match that MTU, either by truncating or splitting logging packets. This contrasts to the superclass, which will simply drop packets that exceed the MTU (optionally logging an error about the failure).

Notes:

The maximum UDP packet header in bytes is 28 bytes (20 bytes IP header + 8 bytes UDP header). Note that the optional 40-byte IP header options could make this 68 bytes, but this is rarely used, and this handler does not use it.

The Syslog priority and facility are encoded into a single 32-bit (4-byte) value.

OVERFLOW_BEHAVIOR_FRAGMENT = 0

OVERFLOW_BEHAVIOR_TRUNCATE = 1

__init__(address=('localhost', 514), facility=1, socket_type=None, overflow=0)

Initialize a handler.

If address is specified as a string, a UNIX socket is used. To log to a local syslogd, “SysLogHandler(address=”/dev/log”)” can be used. If facility is not specified, LOG_USER is used. If socktype is specified as socket.SOCK_DGRAM or socket.SOCK_STREAM, that specific socket type will be used. For Unix sockets, you can also specify a socktype of None, in which case socket.SOCK_DGRAM will be used, falling back to socket.SOCK_STREAM.

emit(record: <class 'logging.LogRecord'>) → None

Emits a record. The record is sent carefully, according to the following rules, to ensure that data is not lost by exceeding the MTU of the connection.

• If the byte-encoded record length plus prefix length plus suffix length plus priority length is less than the maximum allowed length, then a single packet is sent, containing the priority, prefix, full record, and suffix, in that order.

• If it’s greater than or equal to the maximum allowed length and the overflow behavior is set to “truncate,” the record is cleanly truncated (being careful not to split in the middle of a multi-byte character), and then a single packet is sent, containing the priority, prefix, truncated record, and suffix, in that order.

• If it’s greater than or equal to the maximum allowed length and the overflow behavior is set to “fragment,” the record preamble (things like file name, logger name, correlation ID, etc.) is extracted from the start of the record to calculate a new chunk length. The remainder of the record (which should just be the true message and any exception info) is then chunked (being careful not to split in the middle of a multi-byte character) into lengths less than or equal to the chunk length, and then the record is sent as multiple packets, each packet containing the priority, prefix, record preamble, message chunk, and suffix, in that order.

class pysoa.server.coroutine.CoroutineMiddleware

Bases: object

A special middleware that can be used to wrap the execution of coroutines invoked with EnrichedActionRequest.run_coroutine.

before_run_coroutine() → None

When request.run_coroutine is called, this method will be invoked, synchronously, in the calling context (e.g., the calling thread), to execute any logic necessary before handing the coroutine off to the async event loop.

The advantage of implementing this method instead of or in addition to implementing coroutine is that, when multiple middleware are configured, this method will be called in the same order the middleware are configured, while coroutine will be called in the reverse order the middleware are configured (in order to create a coroutine call stack in the same order the middleware are configured).

coroutine(coroutine: MiddlewareCoroutine) → MiddlewareCoroutine -> ~MiddlewareCoroutine

Returns a coroutine (async def ...) that wraps the given coroutine. This wrapping coroutine can be used to execute code before and after the target coroutine. The wrapping pattern is identical to server and client middleware except that it deals with coroutines instead of callables. The wrapping coroutine should return the value it awaits from the coroutine it wraps.

Example:

class CustomCoroutineMiddleware:
    ...

    def coroutine(self, coroutine: MiddlewareCoroutine) -> MiddlewareCoroutine:
        async def wrapper():
            do_stuff_before_coroutine()

            try:
                return await coroutine
            finally:
                do_stuff_after_coroutine()

        return wrapper()

Note: When multiple middleware are configured, this method will be called in the reverse order the middleware are configured, in order to create a coroutine call stack in the same order the middleware are configured. For example, if Middleware1 and Middleware2 are configured in that order, Middleware2.coroutine() will be called before Middleware1.coroutine(), so that the coroutine call stack will be middleware_wrapper_1 -> middleware_wrapper_2 -> coroutine.

Parameters

coroutine – The target coroutine (or coroutine returned by the previous middleware)

Returns

The new, wrapping coroutine (or the unmodified coroutine argument, if no wrapping necessary).

class pysoa.server.coroutine.DefaultCoroutineMiddleware

Bases: pysoa.server.coroutine.CoroutineMiddleware

Your server should have this middleware configured as the very first coroutine middleware. All of your custom coroutine middleware should be configured after this.

Class Configuration Schema

strict dict: The default coroutine middleware has no constructor arguments

No keys permitted.

coroutine(coroutine: MiddlewareCoroutine) → MiddlewareCoroutine -> ~MiddlewareCoroutine

Returns a coroutine (async def ...) that wraps the given coroutine. This wrapping coroutine can be used to execute code before and after the target coroutine. The wrapping pattern is identical to server and client middleware except that it deals with coroutines instead of callables. The wrapping coroutine should return the value it awaits from the coroutine it wraps.

Example:

class CustomCoroutineMiddleware:
    ...

    def coroutine(self, coroutine: MiddlewareCoroutine) -> MiddlewareCoroutine:
        async def wrapper():
            do_stuff_before_coroutine()

            try:
                return await coroutine
            finally:
                do_stuff_after_coroutine()

        return wrapper()

Note: When multiple middleware are configured, this method will be called in the reverse order the middleware are configured, in order to create a coroutine call stack in the same order the middleware are configured. For example, if Middleware1 and Middleware2 are configured in that order, Middleware2.coroutine() will be called before Middleware1.coroutine(), so that the coroutine call stack will be middleware_wrapper_1 -> middleware_wrapper_2 -> coroutine.

Parameters

coroutine – The target coroutine (or coroutine returned by the previous middleware)

Returns

The new, wrapping coroutine (or the unmodified coroutine argument, if no wrapping necessary).

pysoa.server.coroutine.MiddlewareCoroutine = ~MiddlewareCoroutine

A TypeVar for typing middleware coroutine methods to ensure the wrapped and wrapping coroutines have the same return value.

exception pysoa.server.errors.ActionError(errors, set_is_caller_error_to=True)

Bases: pysoa.server.errors.PySOAServerError

Raised by action code, middleware, or the server class as a flow control mechanism for returning an pysoa.common.types.ActionResponse with at least one Error in it.

__init__(errors: List[pysoa.common.errors.Error], set_is_caller_error_to: Optional[bool] = True) → None

Constructs a new action error.

Parameters

• errors – The list of Error objects associated with this action error.

• set_is_caller_error_to – If non-None, all of the Error objects in errors will have their is_caller_error attribute set to this value. Defaults to True, so you should set this to None if you do not desire the input errors to be modified.

exception pysoa.server.errors.JobError(errors, set_is_caller_error_to=False)

Bases: pysoa.server.errors.PySOAServerError

Raised by middleware or the server class as a flow control mechanism for returning a pysoa.common.types.JobResponse with at least one Error in it.

__init__(errors: List[pysoa.common.errors.Error], set_is_caller_error_to: Optional[bool] = False) → None

Constructs a new job error.

Parameters

• errors – The list of Error objects associated with this job error.

• set_is_caller_error_to – If non-None, all of the Error objects in errors will have their is_caller_error attribute set to this value. Defaults to False, so you should set this to None if you do not desire the input errors to be modified.

exception pysoa.server.errors.PySOAServerError

Bases: pysoa.common.errors.PySOAError

Base exception for all server-side errors other than transport errors.

exception pysoa.server.errors.ResponseValidationError(action, errors)

Bases: pysoa.server.errors.PySOAServerError

Raised by an action when the response fails to validate against the defined response schema for that action. Indicates a server-side programming error that must be corrected.

__init__(action: str, errors: List[conformity.types.Error]) → None

Initialize self. See help(type(self)) for accurate signature.

__str__()

Return str(self).

class pysoa.server.middleware.ServerMiddleware

Bases: object

Base middleware class for server middleware. Not required, but provides some helpful stubbed methods and documentation that you should follow for creating your middleware classes. If you extend this class, you may override either one or both of the methods.

Middleware must have two callable attributes, job and action, that, when called with the next level down, return a callable that takes the appropriate arguments and returns the appropriate value.

Class Configuration Schema

strict dict: Most server middleware has no constructor arguments, but subclasses can override this schema

No keys permitted.

action(process_action: Callable[[pysoa.server.types.EnrichedActionRequest], pysoa.common.types.ActionResponse]) → Callable[[pysoa.server.types.EnrichedActionRequest], pysoa.common.types.ActionResponse]

In sub-classes, used for creating a wrapper around process_action. In this simple implementation, just returns process_action.

Parameters

process_action – A callable that accepts a pysoa.server.types.EnrichedActionRequest object and returns a pysoa.common.types.ActionResponse object, or raises an exception.

Returns

A callable that accepts a pysoa.server.types.EnrichedActionRequest object and returns a pysoa.common.types.ActionResponse object, or raises an exception, by calling the provided process_action argument and possibly doing other things.

job(process_job: Callable[[pysoa.server.types.EnrichedJobRequest], pysoa.common.types.JobResponse]) → Callable[[pysoa.server.types.EnrichedJobRequest], pysoa.common.types.JobResponse]

In sub-classes, used for creating a wrapper around process_job. In this simple implementation, just returns ‘process_job`.

Parameters

process_job – A callable that accepts a pysoa.server.types.EnrichedJobRequest and returns a pysoa.common.types.JobResponse object, or raises an exception.

Returns

A callable that accepts a pysoa.server.types.EnrichedJobRequest and returns a pysoa.common.types.JobResponse object, or raises an exception, by calling the provided process_job argument and possibly doing other things.

ActionRequestSchema = pre-defined Conformity schema pysoa.server.schemas.ActionRequestSchema

The Conformity schema with which action requests are validated.

strict dict: (no description)

• action - unicode: The name of the service action to execute.

• body - flexible dict: The request parameters for this action.

  keys

  unicode: (no description)

  values

  anything: (no description)

Optional keys: body

JobRequestSchema = pre-defined Conformity schema pysoa.server.schemas.JobRequestSchema

The Conformity schema with which job requests are validated.

strict dict: (no description)

• actions - list: The list of all actions to execute in this request

  values

  strict dict: (no description)

  • action - unicode: The name of the service action to execute.

  • body - flexible dict: The request parameters for this action.

    keys

    unicode: (no description)

    values

    anything: (no description)

  Optional keys: body

• context - strict dict: (no description)

  • caller - unicode: Optional, caller-supplied meta-information about the caller, such as an application name, file name, class or method name, file name and line number, etc. May be useful for logging or metrics.

  • calling_service - unicode: A header that PySOA automatically adds to all outgoing requests from one service to another when those requests are made with request.client.call_**(...) or request.client.send_request(...). For example, if Foo Service calls Bar Service, the request context that Bar Service receives will include "calling_service": "foo". May be useful for logging or metrics.

  • correlation_id - unicode: Correlation IDs can be used at your own discretion, but are generally shared across multiple service requests, even across multiple services, to correlate requests that are logically linked together (example: such as all PySOA requests that occur within the scope of a single HTTP request in a client application). The PySOA client automatically adds a UUID correlation ID to all outgoing requests if the client is not already configured with an inherited correlation ID, and the client available in request.client automatically inherits the correlation ID from the request.

  • switches - list: See: Versioning using switches.

    values

    integer: (no description)

  Extra keys of any value are allowed. Optional keys: caller, calling_service

• control - strict dict: (no description)

  • continue_on_error - boolean: Whether to continue executing more actions in a multi-action job request if an action results in an error.

  • suppress_response - boolean: Whether to complete processing a request without sending a response back to the client (defaults to false).

  Extra keys of any value are allowed. Optional keys: suppress_response

class pysoa.server.settings.ServerSettings(data)

Bases: pysoa.common.settings.SOASettings

Base settings class for all servers, whose middleware values are restricted to subclasses of ServerMiddleware and whose transport values are restricted to subclasses of BaseServerTransport. Middleware and transport configuration settings schemas will automatically switch based on the configuration settings schema for the path for each.

Settings Schema Definition

• client_routing - flexible dict: Client settings for sending requests to other services; keys should be service names, and values should be the corresponding configuration dicts, which will be validated using the ClientSettings schema.

  keys

  unicode: (no description)

  values

  flexible dict: (no description)

  keys

  hashable: (no description)

  values

  anything: (no description)

• coroutine_middleware - list: The list of all CoroutineMiddleware classes that should be constructed and applied to request.run_coroutine calls processed by this server. By default, pysoa.server.coroutine:DefaultCoroutineMiddleware will be configured first. You can change and/or add to this, but we recommend that you always configure DefaultCoroutineMiddleware as the first middleware.

  values

  dictionary with keys path and kwargs whose kwargs schema switches based on the value of path, dynamically based on class imported from path (see the configuration settings schema documentation for the class named at path). The imported item at the specified path must be a subclass of pysoa.server.coroutine.CoroutineMiddleware.

• extra_fields_to_redact - set: Use this field to supplement the set of fields that are automatically redacted/censored in request and response fields with additional fields that your service needs redacted.

  values

  unicode: (no description)

• harakiri - strict dict: Instructions for automatically terminating a server process when request processing takes longer than expected.

  • shutdown_grace - integer: Seconds to forcefully shutdown after harakiri is triggered if shutdown does not occur (additional information: {'gt': 0})

  • timeout - integer: Seconds of inactivity before harakiri is triggered; 0 to disable, defaults to 300 (additional information: {'gte': 0})

• heartbeat_file - unicode (nullable): If specified, the server will create a heartbeat file at the specified path on startup, update the timestamp in that file after the processing of every request or every time idle operations are processed, and delete the file when the server shuts down. The file name can optionally contain the specifier {{pid}}, which will be replaced with the server process PID. Finally, the file name can optionally contain the specifier {{fid}}, which will be replaced with the unique-and-deterministic forked process ID whenever the server is started with the –fork option (the minimum value is always 1 and the maximum value is always equal to the value of the –fork option).

• logging - strict dict: Settings to enforce the standard Python logging dictionary-based configuration, as you would load with logging.config.dictConfig(). For more information than the documentation here, see https://docs.python.org/3/library/logging.config.html#configuration-dictionary-schema.

  • version - integer: (no description) (additional information: {'gte': 1, 'lte': 1})

  • formatters - flexible dict: This defines a mapping of logging formatter names to formatter configurations. The format key specifies the log format and the datefmt key specifies the date format.

    keys

    unicode: (no description)

    values

    strict dict: (no description)

    • datefmt - unicode: The optional date format used when formatting dates in the log output (see https://docs.python.org/3/library/datetime.html#strftime-strptime-behavior).

    • format - unicode: The format string for this formatter (see https://docs.python.org/3/library/logging.html#logrecord-attributes).

    Optional keys: datefmt

  • filters - flexible dict: This defines a mapping of logging filter names to filter configurations. If a config has only the name key, then logging.Filter will be instantiated with that argument. You can specify a () key (yes, really) to override the default logging.Filter class with a custom filter implementation (which should extend logging.Filter). Extra keys are allowed only for custom implementations having extra constructor arguments matching those key names.

    keys

    unicode: (no description)

    values

    strict dict: (no description)

    • () - a unicode string importable Python path in the format “foo.bar.MyClass”, “foo.bar:YourClass.CONSTANT”, etc. The optional, fully-qualified name of the class extending logging.Filter, used to override the default class logging.Filter. The imported item at the specified path must match the following schema:

      schema

      a Python type that is a subclass of the following class or classes: logging.Filter.

    • name - unicode: The optional filter name which will be passed to the name argument of the logging.Filter class.

    Extra keys of any value are allowed. Optional keys: (), name

  • handlers - flexible dict: This defines a mapping of logging handler names to handler configurations. The class key is the importable Python path to the class extending logging.Handler. The level and filters keys apply to all handlers. The formatter key is valid for all handlers, but not all handlers will use it. Extra keys are allowed only for handlers having extra constructor arguments matching those key names.

    keys

    unicode: (no description)

    values

    strict dict: (no description)

    • class - a unicode string importable Python path in the format “foo.bar.MyClass”, “foo.bar:YourClass.CONSTANT”, etc. The fully-qualified name of the class extending logging.Handler. The imported item at the specified path must match the following schema:

      schema

      a Python type that is a subclass of the following class or classes: logging.Handler.

    • filters - list: A list of references to keys from filters for assigning those filters to this handler.

      values

      unicode: (no description)

    • formatter - unicode: A reference to a key from formatters for assigning that formatter to this handler.

    • level - constant: The logging level at or above which this handler will emit logging events. (additional information: {'values': ['CRITICAL', 'DEBUG', 'ERROR', 'INFO', 'WARNING']})

    Extra keys of any value are allowed. Optional keys: filters, formatter, level

  • loggers - flexible dict: This defines a mapping of logger names to logger configurations. A log event not handled by one of these configured loggers (if any) will instead be handled by the root logger. A log event handled by one of these configured loggers may still be handled by another logger or the root logger unless its propagate key is set to False.

    keys

    unicode: (no description)

    values

    strict dict: (no description)

    • filters - list: A list of references to keys from filters for assigning those filters to this logger.

      values

      unicode: (no description)

    • handlers - list: A list of references to keys from handlers for assigning those handlers to this logger.

      values

      unicode: (no description)

    • level - constant: The logging level at or above which this logger will handle logging events and send them to its configured handlers. (additional information: {'values': ['CRITICAL', 'DEBUG', 'ERROR', 'INFO', 'WARNING']})

    • propagate - boolean: Whether logging events handled by this logger should propagate to other loggers and/or the root logger. Defaults to True.

    Optional keys: filters, handlers, level, propagate

  • root - strict dict: (no description)

    • filters - list: A list of references to keys from filters for assigning those filters to this logger.

      values

      unicode: (no description)

    • handlers - list: A list of references to keys from handlers for assigning those handlers to this logger.

      values

      unicode: (no description)

    • level - constant: The logging level at or above which this logger will handle logging events and send them to its configured handlers. (additional information: {'values': ['CRITICAL', 'DEBUG', 'ERROR', 'INFO', 'WARNING']})

    Optional keys: filters, handlers, level

  • incremental - boolean: Whether this configuration should be considered incremental to any existing configuration. It defaults to False and it is rare that you should ever need to change that.

  • disable_existing_loggers - boolean: Whether all existing loggers (objects obtained from logging.getLogger()) should be disabled when this logging config is loaded. Take our advice and always set this to False. It defaults to True and you almost never want that, because loggers in already-loaded modules will stop working.

  Optional keys: disable_existing_loggers, filters, formatters, handlers, incremental, loggers, root, version

• metrics - dictionary with keys path and kwargs whose kwargs schema switches based on the value of path, dynamically based on class imported from path (see the configuration settings schema documentation for the class named at path). Configuration for defining a usage and performance metrics recorder. The imported item at the specified path must be a subclass of pymetrics.recorders.base.MetricsRecorder.

• middleware - list: The list of all ServerMiddleware objects that should be applied to requests processed by this server

  values

  dictionary with keys path and kwargs whose kwargs schema switches based on the value of path, dynamically based on class imported from path (see the configuration settings schema documentation for the class named at path). The imported item at the specified path must be a subclass of pysoa.server.middleware.ServerMiddleware.

• request_log_error_level - constant: The logging level at which full request and response contents will be logged for requests whose responses contain errors (setting this to a more severe level than request_log_success_level will allow you to easily filter for unsuccessful requests) (additional information: {'values': ['CRITICAL', 'DEBUG', 'ERROR', 'INFO', 'WARNING']})

• request_log_success_level - constant: The logging level at which full request and response contents will be logged for successful requests (additional information: {'values': ['CRITICAL', 'DEBUG', 'ERROR', 'INFO', 'WARNING']})

• transport - dictionary with keys path and kwargs whose kwargs schema switches based on the value of path, dynamically based on class imported from path (see the configuration settings schema documentation for the class named at path). The imported item at the specified path must be a subclass of pysoa.common.transport.base.ServerTransport.

Default Values

Keys present in the dict below can be omitted from compliant settings dicts, in which case the values below will apply as the default values.

{
    "client_routing": {},
    "coroutine_middleware": [
        {
            "path": "pysoa.server.coroutine:DefaultCoroutineMiddleware"
        }
    ],
    "extra_fields_to_redact": [],
    "harakiri": {
        "shutdown_grace": 30,
        "timeout": 300
    },
    "heartbeat_file": null,
    "logging": {
        "disable_existing_loggers": false,
        "filters": {
            "pysoa_logging_context_filter": {
                "()": "pysoa.common.logging.PySOALogContextFilter"
            }
        },
        "formatters": {
            "console": {
                "format": "%(asctime)s %(levelname)7s %(correlation_id)s %(request_id)s: %(message)s"
            },
            "syslog": {
                "format": "%(service_name)s_service: %(name)s %(levelname)s %(module)s %(process)d correlation_id %(correlation_id)s request_id %(request_id)s %(message)s"
            }
        },
        "handlers": {
            "console": {
                "class": "logging.StreamHandler",
                "filters": [
                    "pysoa_logging_context_filter"
                ],
                "formatter": "console",
                "level": "INFO"
            },
            "syslog": {
                "address": [
                    "localhost",
                    514
                ],
                "class": "pysoa.common.logging.SyslogHandler",
                "facility": 23,
                "filters": [
                    "pysoa_logging_context_filter"
                ],
                "formatter": "syslog",
                "level": "INFO"
            }
        },
        "loggers": {},
        "root": {
            "handlers": [
                "console"
            ],
            "level": "INFO"
        },
        "version": 1
    },
    "metrics": {
        "path": "pymetrics.recorders.noop:NonOperationalMetricsRecorder"
    },
    "middleware": [],
    "request_log_error_level": "INFO",
    "request_log_success_level": "INFO",
    "transport": {
        "path": "pysoa.common.transport.redis_gateway.server:RedisServerTransport"
    }
}

defaults = {'client_routing': {}, 'coroutine_middleware': [{'path': 'pysoa.server.coroutine:DefaultCoroutineMiddleware'}], 'extra_fields_to_redact': {}, 'harakiri': {'shutdown_grace': 30, 'timeout': 300}, 'heartbeat_file': None, 'logging': {'disable_existing_loggers': False, 'filters': {'pysoa_logging_context_filter': {'()': 'pysoa.common.logging.PySOALogContextFilter'}}, 'formatters': {'console': {'format': '%(asctime)s %(levelname)7s %(correlation_id)s %(request_id)s: %(message)s'}, 'syslog': {'format': '%(service_name)s_service: %(name)s %(levelname)s %(module)s %(process)d correlation_id %(correlation_id)s request_id %(request_id)s %(message)s'}}, 'handlers': {'console': {'class': 'logging.StreamHandler', 'filters': ['pysoa_logging_context_filter'], 'formatter': 'console', 'level': 'INFO'}, 'syslog': {'address': ('localhost', 514), 'class': 'pysoa.common.logging.SyslogHandler', 'facility': 23, 'filters': ['pysoa_logging_context_filter'], 'formatter': 'syslog', 'level': 'INFO'}}, 'loggers': {}, 'root': {'handlers': ['console'], 'level': 'INFO'}, 'version': 1}, 'metrics': {'path': 'pymetrics.recorders.noop:NonOperationalMetricsRecorder'}, 'middleware': [], 'request_log_error_level': 'INFO', 'request_log_success_level': 'INFO', 'transport': {'path': 'pysoa.common.transport.redis_gateway.server:RedisServerTransport'}}

schema = {'client_routing': SchemalessDictionary(key_type=UnicodeString(min_length=None, max_length=None, description=None, allow_blank=True), value_type=SchemalessDictionary(key_type=Hashable(description=None), value_type=Anything(description=None), max_length=None, min_length=None, description=None, additional_validator=None), max_length=None, min_length=None, description='Client settings for sending requests to other services; keys should be service names, and values should be the corresponding configuration dicts, which will be validated using the ClientSettings schema.', additional_validator=None), 'coroutine_middleware': List(contents=ClassConfigurationSchema(base_class=<class 'pysoa.server.coroutine.CoroutineMiddleware'>, default_path=None, description=None, eager_default_validation=True, add_class_object_to_dict=True), max_length=None, min_length=None, description='The list of all `CoroutineMiddleware` classes that should be constructed and applied to `request.run_coroutine` calls processed by this server. By default, `pysoa.server.coroutine:DefaultCoroutineMiddleware` will be configured first. You can change and/or add to this, but we recommend that you always configure `DefaultCoroutineMiddleware` as the first middleware.', additional_validator=None), 'extra_fields_to_redact': Set(contents=UnicodeString(min_length=None, max_length=None, description=None, allow_blank=True), max_length=None, min_length=None, description='Use this field to supplement the set of fields that are automatically redacted/censored in request and response fields with additional fields that your service needs redacted.', additional_validator=None), 'harakiri': Dictionary(contents={'timeout': Integer(gt=None, gte=0, lt=None, lte=None, description='Seconds of inactivity before harakiri is triggered; 0 to disable, defaults to 300'), 'shutdown_grace': Integer(gt=0, gte=None, lt=None, lte=None, description='Seconds to forcefully shutdown after harakiri is triggered if shutdown does not occur')}, optional_keys=frozenset(), allow_extra_keys=False, description='Instructions for automatically terminating a server process when request processing takes longer than expected.', additional_validator=None), 'heartbeat_file': Nullable(field=UnicodeString(min_length=None, max_length=None, description='If specified, the server will create a heartbeat file at the specified path on startup, update the timestamp in that file after the processing of every request or every time idle operations are processed, and delete the file when the server shuts down. The file name can optionally contain the specifier {{pid}}, which will be replaced with the server process PID. Finally, the file name can optionally contain the specifier {{fid}}, which will be replaced with the unique-and-deterministic forked process ID whenever the server is started with the --fork option (the minimum value is always 1 and the maximum value is always equal to the value of the --fork option).', allow_blank=True)), 'logging': Dictionary(contents=OrderedDict([('version', Integer(gt=None, gte=1, lt=None, lte=1, description=None)), ('formatters', SchemalessDictionary(key_type=UnicodeString(min_length=None, max_length=None, description=None, allow_blank=True), value_type=Dictionary(contents={'format': UnicodeString(min_length=None, max_length=None, description='The format string for this formatter (see https://docs.python.org/3/library/logging.html#logrecord-attributes).', allow_blank=True), 'datefmt': UnicodeString(min_length=None, max_length=None, description='The optional date format used when formatting dates in the log output (see https://docs.python.org/3/library/datetime.html#strftime-strptime-behavior).', allow_blank=True)}, optional_keys=frozenset({'datefmt'}), allow_extra_keys=False, description=None, additional_validator=None), max_length=None, min_length=None, description='This defines a mapping of logging formatter names to formatter configurations. The `format` key specifies the log format and the `datefmt` key specifies the date format.', additional_validator=None)), ('filters', SchemalessDictionary(key_type=UnicodeString(min_length=None, max_length=None, description=None, allow_blank=True), value_type=Dictionary(contents={'()': TypePath(value_schema=TypeReference(base_classes=<class 'logging.Filter'>, description=None), description='The optional, fully-qualified name of the class extending `logging.Filter`, used to override the default class `logging.Filter`.'), 'name': UnicodeString(min_length=None, max_length=None, description='The optional filter name which will be passed to the `name` argument of the `logging.Filter` class.', allow_blank=True)}, optional_keys=frozenset({'name', '()'}), allow_extra_keys=True, description=None, additional_validator=None), max_length=None, min_length=None, description='This defines a mapping of logging filter names to filter configurations. If a config has only the `name` key, then `logging.Filter` will be instantiated with that argument. You can specify a `()` key (yes, really) to override the default `logging.Filter` class with a custom filter implementation (which should extend `logging.Filter`). Extra keys are allowed only for custom implementations having extra constructor arguments matching those key names.', additional_validator=None)), ('handlers', SchemalessDictionary(key_type=UnicodeString(min_length=None, max_length=None, description=None, allow_blank=True), value_type=Dictionary(contents={'class': TypePath(value_schema=TypeReference(base_classes=<class 'logging.Handler'>, description=None), description='The fully-qualified name of the class extending `logging.Handler`.'), 'level': PythonLogLevel(), 'formatter': UnicodeString(min_length=None, max_length=None, description='A reference to a key from `formatters` for assigning that formatter to this handler.', allow_blank=True), 'filters': List(contents=UnicodeString(min_length=None, max_length=None, description=None, allow_blank=True), max_length=None, min_length=None, description='A list of references to keys from `filters` for assigning those filters to this handler.', additional_validator=None)}, optional_keys=frozenset({'formatter', 'filters', 'level'}), allow_extra_keys=True, description=None, additional_validator=None), max_length=None, min_length=None, description='This defines a mapping of logging handler names to handler configurations. The `class` key is the importable Python path to the class extending `logging.Handler`. The `level` and `filters` keys apply to all handlers. The `formatter` key is valid for all handlers, but not all handlers will use it. Extra keys are allowed only for handlers having extra constructor arguments matching those key names.', additional_validator=None)), ('loggers', SchemalessDictionary(key_type=UnicodeString(min_length=None, max_length=None, description=None, allow_blank=True), value_type=Dictionary(contents={'level': PythonLogLevel(), 'filters': List(contents=UnicodeString(min_length=None, max_length=None, description=None, allow_blank=True), max_length=None, min_length=None, description='A list of references to keys from `filters` for assigning those filters to this logger.', additional_validator=None), 'handlers': List(contents=UnicodeString(min_length=None, max_length=None, description=None, allow_blank=True), max_length=None, min_length=None, description='A list of references to keys from `handlers` for assigning those handlers to this logger.', additional_validator=None), 'propagate': Boolean(description='Whether logging events handled by this logger should propagate to other loggers and/or the root logger. Defaults to `True`.')}, optional_keys=frozenset({'filters', 'handlers', 'level', 'propagate'}), allow_extra_keys=False, description=None, additional_validator=None), max_length=None, min_length=None, description='This defines a mapping of logger names to logger configurations. A log event not handled by one of these configured loggers (if any) will instead be handled by the root logger. A log event handled by one of these configured loggers may still be handled by another logger or the root logger unless its `propagate` key is set to `False`.', additional_validator=None)), ('root', Dictionary(contents={'level': PythonLogLevel(), 'filters': List(contents=UnicodeString(min_length=None, max_length=None, description=None, allow_blank=True), max_length=None, min_length=None, description='A list of references to keys from `filters` for assigning those filters to this logger.', additional_validator=None), 'handlers': List(contents=UnicodeString(min_length=None, max_length=None, description=None, allow_blank=True), max_length=None, min_length=None, description='A list of references to keys from `handlers` for assigning those handlers to this logger.', additional_validator=None)}, optional_keys=frozenset({'filters', 'handlers', 'level'}), allow_extra_keys=False, description=None, additional_validator=None)), ('incremental', Boolean(description='Whether this configuration should be considered incremental to any existing configuration. It defaults to `False` and it is rare that you should ever need to change that.')), ('disable_existing_loggers', Boolean(description='Whether all existing loggers (objects obtained from `logging.getLogger()`) should be disabled when this logging config is loaded. Take our advice and *always* set this to `False`. It defaults to `True` and you almost never want that, because loggers in already-loaded modules will stop working.'))]), optional_keys=frozenset({'incremental', 'version', 'filters', 'loggers', 'disable_existing_loggers', 'root', 'formatters', 'handlers'}), allow_extra_keys=False, description='Settings to enforce the standard Python logging dictionary-based configuration, as you would load with `logging.config.dictConfig()`. For more information than the documentation here, see https://docs.python.org/3/library/logging.config.html#configuration-dictionary-schema.', additional_validator=<conformity.fields.logging._LoggingValidator object>), 'metrics': ClassConfigurationSchema(base_class=<class 'pymetrics.recorders.base.MetricsRecorder'>, default_path=None, description='Configuration for defining a usage and performance metrics recorder.', eager_default_validation=True, add_class_object_to_dict=True), 'middleware': List(contents=ClassConfigurationSchema(base_class=<class 'pysoa.server.middleware.ServerMiddleware'>, default_path=None, description=None, eager_default_validation=True, add_class_object_to_dict=True), max_length=None, min_length=None, description='The list of all `ServerMiddleware` objects that should be applied to requests processed by this server', additional_validator=None), 'request_log_error_level': PythonLogLevel(), 'request_log_success_level': PythonLogLevel(), 'transport': ClassConfigurationSchema(base_class=<class 'pysoa.common.transport.base.ServerTransport'>, default_path=None, description=None, eager_default_validation=True, add_class_object_to_dict=True)}

exception pysoa.server.server.HarakiriInterrupt

Bases: BaseException

Raised internally to notify the server code about interrupts due to harakiri. You should never, ever, ever, ever catch this exception in your service code. As such, it inherits from BaseException so that even except Exception: won’t catch it. However, except: will catch it, so, per standard Python coding standards, you should never use except: (or except BaseException:, for that matter).

class pysoa.server.server.Server(settings, forked_process_id=None)

Bases: object

The base class from which all PySOA service servers inherit, and contains the code that does all of the heavy lifting for receiving and handling requests, passing those requests off to the relevant actions, and sending the actions’ responses back to the caller.

Required attributes that all concrete subclasses must provide:

• service_name: A (unicode) string name of the service.

• action_class_map: An object supporting __contains__ and __getitem__ (typically a dict) whose keys are action names and whose values are callable objects that return a callable action when called (such as subclasses of Action which, when “called” [constructed], yield a callable object [instance of the subclass])

__init__(settings: <class 'pysoa.server.settings.ServerSettings'>, forked_process_id: Optional[int] = None) → None

Parameters

• settings – The settings object, which must be an instance of ServerSettings or one of its subclasses

• forked_process_id – If multiple processes are forked by the same parent process, this will be set to a unique, deterministic (incremental) ID which can be used in logging, the heartbeat file, etc. For example, if the --fork argument is used with the value 5 (creating five child processes), this argument will have the values 1, 2, 3, 4, and 5 across the five respective child processes.

action_class_map = {}

client_class

alias of pysoa.client.client.Client

execute_job(job_request: <class 'pysoa.server.types.EnrichedJobRequest'>) → <class 'pysoa.common.types.JobResponse'>

Processes and runs the action requests contained in the job and returns a JobResponse.

Parameters

job_request – The job request

Returns

A JobResponse object

handle_job_error_code(code: str, message: str, request_for_logging: <class 'pysoa.common.logging.RecursivelyCensoredDictWrapper'>, response_for_logging: <class 'pysoa.common.logging.RecursivelyCensoredDictWrapper'>, extra: Union[Dict[str, Any], None] = None) → <class 'pysoa.common.types.JobResponse'>

Makes and returns a last-ditch error response based on a known, expected (though unwanted) error while logging details about it.

Parameters

• code – The error code.

• message – The error message.

• request_for_logging – The censor-wrapped request dictionary.

• response_for_logging – The censor-wrapped response dictionary.

• extra – Any extra items to add to the logged error.

Returns

A JobResponse object.

handle_next_request() → None

Retrieves the next request from the transport, or returns if it times out (no request has been made), and then processes that request, sends its response, and returns when done.

handle_shutdown_signal(signal_number: int, _stack_frame: <class 'frame'>) → None

Handles the reception of a shutdown signal.

handle_unhandled_exception(exception: <class 'Exception'>, response_type: Type[~_RT], variables: Union[Dict[str, Any], None] = None, **kwargs: Any) → ~_RT

Makes and returns a last-ditch error response based on an unknown, unexpected error.

Parameters

• exception – The exception that happened.

• response_type – The response type (JobResponse or ActionResponse) that should be created.

• variables – An optional dictionary of context-relevant variables to include in the error response.

• kwargs – Keyword arguments that will be passed to the response object created.

Returns

A JobResponse object or ActionResponse error based on the response_type argument.

harakiri(signal_number: int, _stack_frame: <class 'frame'>) → None

Handles the reception of a timeout signal indicating that a request has been processing for too long, as defined by the harakiri settings. This method makes use of two “private” Python functions, sys._current_frames and os._exit, but both of these functions are publicly documented and supported.

classmethod initialize(settings: <class 'pysoa.server.settings.ServerSettings'>) → Type[pysoa.server.server.Server]

Called just before the Server class is instantiated, and passed the settings dict. Can be used to perform settings manipulation, server class patching (such as for performance tracing operations), and more. Use with great care and caution. Overriding methods must call super and return cls or a new/modified cls, which will be used to instantiate the server. See the documentation for Server.main for full details on the chain of Server method calls.

Returns

The server class or a new/modified server class

introspection_action = None

classmethod main(forked_process_id: Optional[int] = None) → None

Command-line entry point for running a PySOA server. The chain of method calls is as follows:

cls.main
  |
  -> cls.initialize => new_cls
  -> new_cls.__init__ => self
  -> self.run
      |
      -> self.setup
      -> [async event loop started if Python 3.5+]
      -> [heartbeat file created if configured]
      -> loop: self.handle_next_request while not self.shutting_down
                |
                -> transport.receive_request_message
                -> self.perform_idle_actions (if no request)
                -> self.perform_pre_request_actions
                -> self.process_job
                    |
                    -> middleware(self.execute_job)
                -> transport.send_response_message
                -> self.perform_post_request_actions
      -> self.teardown
      -> [async event loop joined in Python 3.5+; this make take a few seconds to finish running tasks]
      -> [Django resources cleaned up]
      -> [heartbeat file deleted if configured]

Parameters

forked_process_id – If multiple processes are forked by the same parent process, this will be set to a unique, deterministic (incremental) ID which can be used in logging, the heartbeat file, etc. For example, if the --fork argument is used with the value 5 (creating five child processes), this argument will have the values 1, 2, 3, 4, and 5 across the five respective child processes.

make_client(context: Dict[str, Any], extra_context: Union[Dict[str, Any], None] = None, **kwargs: Any) → <class 'pysoa.client.client.Client'>

Gets a Client that will propagate the passed context in order to to pass it down to middleware or Actions. The server code will call this method only with the context argument and no other arguments. Subclasses can override this method and replace its behavior completely or call super to pass extra_context data or keyword arguments that will be passed to the client. The supplied context argument will not be modified in any way (it will be copied); the same promise is not made for the extra_context argument.

Parameters

• context – The context parameter, supplied by the server code when making a client

• extra_context – Extra context information supplied by subclasses as they see fit

• kwargs – Keyword arguments that will be passed as-is to the Client constructor

Returns

A Client configured with this server’s client_routing settings and the supplied context, extra context, and keyword arguments.

static make_middleware_stack(middleware: List[Callable[[~_MT], ~_MT]], base: ~_MT) → ~_MT

Given a list of in-order middleware callable objects middleware and a base function base, chains them together so each middleware is fed the function below, and returns the top level ready to call.

Parameters

• middleware – The middleware stack

• base – The base callable that the lowest-order middleware wraps

Returns

The topmost middleware, which calls the next middleware … which calls the lowest-order middleware, which calls the base callable.

perform_idle_actions() → None

Runs periodically when the server is idle, if it has been too long since it last received a request. Call super().perform_idle_actions() if you override. See the documentation for Server.main for full details on the chain of Server method calls.

perform_post_request_actions() → None

Runs just after the server processes a request. Call super().perform_post_request_actions() if you override. Be sure your purpose for overriding isn’t better met with middleware. See the documentation for Server.main for full details on the chain of Server method calls.

perform_pre_request_actions() → None

Runs just before the server accepts a new request. Call super().perform_pre_request_actions() if you override. Be sure your purpose for overriding isn’t better met with middleware. See the documentation for Server.main for full details on the chain of Server method calls.

classmethod pre_fork() → None

Called only if the –fork argument is used to pre-fork multiple worker processes. In this case, it is called by the parent process immediately after signal handlers are set and immediately before the worker sub-processes are spawned. It is never called again in the life span of the parent process, even if a worker process crashes and gets re-spawned.

process_job(job_request: Dict[str, Any]) → <class 'pysoa.common.types.JobResponse'>

Validate, execute, and run the job request, wrapping it with any applicable job middleware.

Parameters

job_request – The job request dict

Returns

A JobResponse object

request_class

alias of pysoa.server.types.EnrichedActionRequest

run() → None

Starts the server run loop and returns after the server shuts down due to a shutdown-request, Harakiri signal, or unhandled exception. See the documentation for Server.main for full details on the chain of Server method calls.

service_name = None

settings_class

alias of pysoa.server.settings.ServerSettings

setup() → None

Runs just before the server starts, if you need to do one-time loads or cache warming. Call super().setup() if you override. See the documentation for Server.main for full details on the chain of Server method calls.

teardown() → None

Runs just before the server shuts down, if you need to do any kind of clean up (like updating a metrics gauge, etc.). Call super().teardown() if you override. See the documentation for Server.main for full details on the chain of Server method calls.

use_django = False

class pysoa.server.types.ActionInterface(settings=None)

Bases: object

Actions should either be callables that accept a ServerSettings object and return another callable that accepts an EnrichedActionRequest and returns an ActionResponse, or they should inherit from this class and implement its abstract methods. Most actions, however, will simply extend pysoa.server.action.base.Action and implement its interface, which is simpler and easier to use.

abstract __call__(action_request: <class 'pysoa.server.types.EnrichedActionRequest'>) → <class 'pysoa.common.types.ActionResponse'>

Execute the action.

Parameters

action_request – The action request

Returns

The action response

abstract __init__(settings: Optional[pysoa.server.settings.ServerSettings] = None) → None

Constructs a new action class.

Parameters

settings – The Server settings

pysoa.server.types.ActionType = typing.Union[typing.Type[pysoa.server.types.ActionInterface], typing.Callable[[typing.Union[pysoa.server.settings.ServerSettings, NoneType]], typing.Callable[[pysoa.server.types.EnrichedActionRequest], pysoa.common.types.ActionResponse]]]

A type used for annotating attributes and arguments that represent any valid action class or callable.

class pysoa.server.types.EnrichedActionRequest(action, body=NOTHING, switches: <class 'pysoa.server.internal.types.RequestSwitchSet'> = NOTHING, context: Dict[str, Any] = NOTHING, control: Dict[str, Any] = NOTHING, client: <class 'pysoa.client.client.Client'> = None, run_coroutine: Callable[[Coroutine], concurrent.futures._base.Future] = None)

Bases: pysoa.common.types.ActionRequest

The action request object that the Server passes to each Action class that it calls. It contains all the information from ActionRequest, plus some extra information from the JobRequest, a client that can be used to call other services, and a helper for running asyncio coroutines.

Also contains a helper for easily calling other local service actions from within an action.

Services and intermediate libraries can subclass this class and change the Server attribute request_class to their subclass in order to use more-advanced request classes. In order for any new attributes such a subclass provides to be copied by call_local_action, they must be attr.ib attributes with a default value.

Parameters

• switches – The set of all switches included in the request context.

• context – The job request context header dictionary.

• control – The job request control header dictionary.

• client – A Client instance created by the server based on its client_routing setting and the context header included in the current request.

• run_routine – A callable that accepts a coroutine object (a typing.Coroutine or collections.abc.Coroutine depending on the Python version), such as the awaitable value returned by as async def function, to be executed by the server’s configured async thread loop. This callable returns a concurrent.futures.Future, which you can await or ignore if you do not wish to wait on a result.

__init__(action, body=NOTHING, switches: pysoa.server.internal.types.RequestSwitchSet = NOTHING, context: Dict[str, Any] = NOTHING, control: Dict[str, Any] = NOTHING, client: pysoa.client.client.Client = None, run_coroutine: Callable[[Coroutine], concurrent.futures._base.Future] = None)

Initialize self. See help(type(self)) for accurate signature.

call_local_action(action: str, body: Dict[str, Any], raise_action_errors: bool = True, is_caller_error: bool = False, context: Union[Dict[str, Any], None] = None) → <class 'pysoa.common.types.ActionResponse'>

This helper calls another action, locally, that resides on the same service, using the provided action name and body. The called action will receive a copy of this request object with different action and body details.

The use of this helper differs significantly from using the PySOA client to call an action. Notably:

• The configured transport is not involved, so no socket activity or serialization/deserialization takes place.

• PySOA server metrics are not recorded and post-action cleanup activities do not occur.

• No “job request” is ever created or transacted.

• No middleware is executed around this action (though, in the future, we might change this decision and add middleware execution to this helper).

Parameters

• action – The action to call (must exist within the action_class_map from the Server class)

• body – The body to send to the action

• raise_action_errors – If True (the default), all action errors will be raised; otherwise, an ActionResponse containing the errors will be returned.

• is_caller_error – If True (defaults to False), raised action errors will be marked as the responsibility of the caller. Action errors are usually the responsibility of the caller, but the default here is the opposite since the responsibility usually lies in the service that is calling itself and should know better.

• context – If specified, any values in this dictionary will override conflicting values in the cloned context.

Returns

the action response.

Raises

ActionError

class pysoa.server.types.EnrichedJobRequest(control=NOTHING, context=NOTHING, actions=NOTHING, client: <class 'pysoa.client.client.Client'> = None, run_coroutine: Callable[[Coroutine], concurrent.futures._base.Future] = None)

Bases: pysoa.common.types.JobRequest

__init__(control=NOTHING, context=NOTHING, actions=NOTHING, client: pysoa.client.client.Client = None, run_coroutine: Callable[[Coroutine], concurrent.futures._base.Future] = None)

Initialize self. See help(type(self)) for accurate signature.

class pysoa.server.action.base.Action(settings=None)

Bases: pysoa.server.types.ActionInterface

Base class from which most SOA service actions should inherit.

Contains the basic framework for implementing an action:

• Subclass and override run() with the body of your code

• Optionally provide a description attribute, which should be a unicode string and is used to display introspection information for the action.

• Optionally provide request_schema and/or response_schema attributes. These should be Conformity Dictionaries, and are used both to validate the request and response body and to display introspection information for the action.

• Optionally provide a validate() method to do custom validation on the request.

__call__(action_request: <class 'pysoa.server.types.EnrichedActionRequest'>) → <class 'pysoa.common.types.ActionResponse'>

Main entry point for actions from the Server (or potentially from tests). Validates that the request matches the request_schema, then calls validate(), then calls run() if validate() raised no errors, and then validates that the return value from run() matches the response_schema before returning it in an ActionResponse.

Parameters

action_request – The request object

Returns

The response object

Raise

ActionError, ResponseValidationError

__init__(settings: Optional[pysoa.server.settings.ServerSettings] = None) → None

Construct a new action. Concrete classes can override this and define a different interface, but they must still pass the server settings to this base constructor by calling super.

Parameters

settings – The server settings object

description = None

request_schema = None

response_schema = None

abstract run(request: <class 'pysoa.server.types.EnrichedActionRequest'>) → Dict[str, Any]

Override this to perform your business logic, and either return a value abiding by the response_schema or raise an ActionError.

Parameters

request – The request object

Returns

The response body, which should validate according to the response_schema.

Raise

ActionError

validate(request: <class 'pysoa.server.types.EnrichedActionRequest'>) → None

Override this to perform custom validation logic before the run() method is run. Raise ActionError if you find issues, otherwise return (the return value is ignored). If this method raises an error, run() will not be called. You do not have to override this method if you don’t want to perform custom validation or prefer to perform it in run().

Parameters

request – The request object

Raise

ActionError

class pysoa.server.action.introspection.IntrospectionAction(server)

Bases: pysoa.server.action.base.Action

This action returns detailed information about the service’s defined actions and the request and response schemas for each action, along with any documentation defined for the action or for the service itself. It can be passed a single action name to return information limited to that single action. Otherwise, it will return information for all of the service’s actions.

This action will be added to your service on your behalf if you do not define an action with name introspect.

Making your services and actions capable of being introspected is simple. If your server class has a description attribute, that will be the service’s documentation that introspection returns. If your server class does not have this attribute but does have a docstring, introspection will use the docstring. The same rule applies to action classes: Introspection first looks for a description attribute and then uses the docstring, if any. If neither of these are found, the applicable service or action documentation will be done.

Introspection then looks at the request_schema and response_schema attributes for each of your actions, and includes the details about these schemas in the returned information for each action. Be sure you include field descriptions in your schema for the most effective documentation possible.

__init__(server: <class 'pysoa.server.server.Server'>) → None

Construct a new introspection action. Unlike its base class, which accepts a server settings object, this must be passed a Server object, from which it will obtain a settings object. The Server code that calls this action has special handling to address this requirement.

Parameters

server – A PySOA server instance

description = "This action returns detailed information about the service's defined actions and the request and response schemas for each action, along with any documentation defined for the action or for the service itself. It can be passed a single action name to return information limited to that single action. Otherwise, it will return information for all of the service's actions. If an action is a switched action (meaning the action extends `SwitchedAction`, and which action code runs is controlled with SOA switches), multiple action introspection results will be returned for that action, each with a name ending in either `[switch:N]` (where `N` is the switch value) or `[DEFAULT]` for the default action."

request_schema = Dictionary(contents={'action_name': UnicodeString(min_length=1, max_length=None, description='Specify this to limit your introspection to a single action. It will be the only action present in the `actions` response attribute. If the requested action does not exist, an error will be returned.', allow_blank=False)}, optional_keys=frozenset({'action_name'}), allow_extra_keys=False, description=None, additional_validator=None)

response_schema = Dictionary(contents={'documentation': Nullable(field=UnicodeString(min_length=None, max_length=None, description='The documentation for the server, unless `action_name` is specified in the request body, in which case this is omitted.', allow_blank=True)), 'action_names': List(contents=UnicodeString(min_length=None, max_length=None, description='The name of an action.', allow_blank=True), max_length=None, min_length=None, description='An alphabetized list of every action name included in `actions`.', additional_validator=None), 'actions': SchemalessDictionary(key_type=UnicodeString(min_length=None, max_length=None, description='The name of the action.', allow_blank=True), value_type=Dictionary(contents={'documentation': Nullable(field=UnicodeString(min_length=None, max_length=None, description='The documentation for the action', allow_blank=True)), 'request_schema': Nullable(field=Anything(description='A description of the expected request schema, including any documentation specified in the schema definition.')), 'response_schema': Nullable(field=Anything(description='A description of the guaranteed response schema, including any documentation specified in the schema definition.'))}, optional_keys=frozenset(), allow_extra_keys=False, description='A introspection of a single action', additional_validator=None), max_length=None, min_length=None, description='A dict mapping action names to action description dictionaries. This contains details about every action in the service unless `action_name` is specified in the request body, in which case it contains details only for that action.', additional_validator=None)}, optional_keys=frozenset({'documentation'}), allow_extra_keys=False, description=None, additional_validator=None)

run(request: <class 'pysoa.server.types.EnrichedActionRequest'>) → Dict[str, Any]

Introspects all of the actions on the server and returns their documentation.

Parameters

request – The request object

Returns

The response

class pysoa.server.action.status.BaseStatusAction(settings=None)

Bases: pysoa.server.action.base.Action

Standard base action for status checks. Returns health check and version information.

If you want to use the status action use StatusActionFactory(version), passing in the version of your service and, optionally, the build of your service. If you do not specify an action with name status in your server, this will be done on your behalf.

If you want to make a custom status action, subclass this class, make self._version return your service’s version string, self._build optionally return your service’s build string, and add any additional health check methods you desire. Health check methods must start with check_.

Health check methods accept a single argument, the request object (an instance of ActionRequest), and return a list of tuples in the format (is_error, code, description) (or a false-y value if there are no problems):

• is_error: True if this is an error, False if it is a warning.

• code: Invariant string for this error, like “MYSQL_FAILURE”

• description: Human-readable description of the problem, like “Could not connect to host on port 1234”

Health check methods can also write to the self.diagnostics dictionary to add additional data which will be sent back with the response if they like. They are responsible for their own key management in this situation.

This base status action comes with a disabled-by-default health check method named _check_client_settings (the leading underscore disables it), which calls status on all other services that this service is configured to call (using verbose: False, which guarantees no further recursive status checking) and includes those responses in this action’s response. To enable this health check, simply reference it as a new, valid check_ method name, like so:

class MyStatusAction(BaseStatusAction):
    ...
    check_client_settings = BaseStatusAction._check_client_settings

__init__(settings: Optional[pysoa.server.settings.ServerSettings] = None) → None

Constructs a new base status action. Concrete status actions can override this if they want, but must call super.

Parameters

settings – The server settings object

description = 'Returns version info for the service, Python, PySOA, and Conformity. If the service has a build string, that is also returned. If the service has defined additional health check behavior and the `verbose` request attribute is not set to `False`, those additional health checks are performed and returned in the `healthcheck` response attribute. If the `verbose` request attribute is set to `False`, the additional health checks are not performed and `healthcheck` is not included in the response (importantly, the `check_` methods are not invoked).'

request_schema = Dictionary(contents={'verbose': Boolean(description='If specified and False, this instructs the status action to return only the baseline status information (Python, service, PySOA, and other library versions) and omit any of the health check operations (no `healthcheck` attribute will be included in the response). This provides a useful way to obtain the service version very quickly without executing the often time-consuming code necessary for the full health check. It defaults to True, which means "return everything."')}, optional_keys=frozenset({'verbose'}), allow_extra_keys=False, description=None, additional_validator=None)

response_schema = Dictionary(contents={'build': UnicodeString(min_length=None, max_length=None, description='The version build string, if applicable.', allow_blank=True), 'conformity': UnicodeString(min_length=None, max_length=None, description='The version of Conformity in use.', allow_blank=True), 'healthcheck': Dictionary(contents={'warnings': List(contents=Tuple(), max_length=None, min_length=None, description='A list of any warnings encountered during the health checks.', additional_validator=None), 'errors': List(contents=Tuple(), max_length=None, min_length=None, description='A list of any errors encountered during the health checks.', additional_validator=None), 'diagnostics': SchemalessDictionary(key_type=UnicodeString(min_length=None, max_length=None, description=None, allow_blank=True), value_type=Anything(description=None), max_length=None, min_length=None, description='A dictionary containing any additional diagnostic information output by the health check operations.', additional_validator=None)}, optional_keys=frozenset({'warnings', 'diagnostics', 'errors'}), allow_extra_keys=False, description='Information about any additional health check operations performed.', additional_validator=None), 'pysoa': UnicodeString(min_length=None, max_length=None, description='The version of PySOA in use.', allow_blank=True), 'python': UnicodeString(min_length=None, max_length=None, description='The version of Python in use.', allow_blank=True), 'version': UnicodeString(min_length=None, max_length=None, description='The version of the responding service.', allow_blank=True)}, optional_keys=frozenset({'build', 'healthcheck'}), allow_extra_keys=False, description=None, additional_validator=None)

run(request: <class 'pysoa.server.types.EnrichedActionRequest'>) → Dict[str, Any]

Adds version information for Conformity, PySOA, Python, and the service to the response, then scans the class for check_ methods and runs them (unless verbose is False).

Parameters

request – The request object

Returns

The response

pysoa.server.action.status.CheckMethodReturn = typing.Union[typing.Iterable[pysoa.server.action.status.CheckMethodStatus], NoneType]

The return type for all status action check_ methods.

class pysoa.server.action.status.CheckMethodStatus(is_error, code, description)

Bases: tuple

A tuple used to internally represent errors or warnings within a status action.

__getnewargs__()

Return self as a plain tuple. Used by copy and pickle.

static __new__(_cls, is_error, code, description)

Create new instance of CheckMethodStatus(is_error, code, description)

property code

Alias for field number 1

property description

Alias for field number 2

property is_error

Alias for field number 0

pysoa.server.action.status.StatusActionFactory(version: str, build: Optional[str] = None, base_class: Type[pysoa.server.action.status.BaseStatusAction] = <class 'pysoa.server.action.status.BaseStatusAction'>) → Type[pysoa.server.action.status.BaseStatusAction]

A factory for creating a new status action class specific to a service.

Parameters

• version – The service version.

• build – The optional service build identifier.

• base_class – The optional base class, to override BaseStatusAction as the base class.

Returns

A class named StatusAction, extending base_class, with _version and _build properties returning the corresponding version and build input parameters, respectively.

pysoa.server.action.status.make_default_status_action_class(server_class: Type[pysoa.server.server.Server]) → Type[pysoa.server.action.status.BaseStatusAction]

class pysoa.server.action.switched.SwitchedAction(settings=None)

Bases: object

A specialized action that defers to other, concrete actions based on request switches. Subclasses must not override any methods and must override switch_to_action_map. switch_to_action_map should be some iterable object that provides __len__ (such as a tuple [recommended] or list). Its items must be indexable objects that provide __len__ (such as a tuple [recommended] or list) and have exactly two elements.

For each item in switch_to_action_map, the first element must be a switch that provides __int__ (such as an actual integer) or a switch that provides an attribute value which, itself, provides __int__ (or is an int). The second element must be an action, such as an action class (e.g. one that extends Action) or any callable that accepts a server settings object and returns a new callable that, itself, accepts an ActionRequest object and returns an ActionResponse object or raises an ActionError.

switch_to_action_map must have at least two items in it. SwitchedAction will iterate over that list, checking the first element (switch) of each item to see if it is enabled in the request. If it is, the second element (the action) of that item will be deferred to. If it finds no items whose switches are enabled, it will use the very last action in switch_to_action_map. As such, you can treat the last item as a default, and its switch could simply be SwitchedAction.DEFAULT_ACTION (although, this is not required: it could also be a valid switch, and it would still be treated as the default in the case that no other items matched).

Example usage:

class UserActionV1(Action):
    ...

class UserActionV2(Action):
    ...

class UserTransitionAction(SwitchedAction):
    switch_to_action_map = (
        (USER_VERSION_2_ENABLED, UserActionV2),
        (SwitchedAction.DEFAULT_ACTION, UserActionV1),
    )

DEFAULT_ACTION = <pysoa.server.action.switched._DefaultAction object>

__call__(action_request: <class 'pysoa.server.types.EnrichedActionRequest'>) → <class 'pysoa.common.types.ActionResponse'>

Main entry point for actions from the Server (or potentially from tests). Finds the appropriate real action to invoke based on the switches enabled in the request, initializes the action with the server settings, and then calls the action with the request object, returning its response directly.

Parameters

action_request – The request object

Returns

The response object

Raise

ActionError, ResponseValidationError

__init__(settings: Optional[pysoa.server.settings.ServerSettings] = None) → None

Construct a new action. Concrete classes should not override this.

Parameters

settings – The server settings object

get_uninitialized_action(action_request: <class 'pysoa.server.types.EnrichedActionRequest'>) → Union[Type[pysoa.server.types.ActionInterface], Callable[[Optional[pysoa.server.settings.ServerSettings]], Callable[[pysoa.server.types.EnrichedActionRequest], pysoa.common.types.ActionResponse]]]

Get the raw action (such as the action class or the base action callable) without instantiating/calling it, based on the switches in the action request, or the default raw action if no switches were present or no switches matched.

Parameters

action_request – The request object

Returns

The action

switch_to_action_map = ()

class pysoa.server.django.cache.PySOAMemcachedCache(server, params)

Bases: django.core.cache.backends.memcached.MemcachedCache

If you want to use Memcached with the python-memcached library, we recommend you use this Django cache engine, which does not close the socket connections to Memcached unless the server is shutting down. You are free to use your own implementation, of course. You can also use this in combination with other cache engines (from this module or others) in a multi-cache Django configuration.

close(for_shutdown: bool = False, **_kwargs: Any) → None

Only call super().close() if the server is shutting down (not between requests).

Parameters

for_shutdown – If True, the socket connections to Memcached are closed (defaults to False).

class pysoa.server.django.cache.PySOAProcessScopedMemoryCache(name, params)

Bases: django.core.cache.backends.locmem.LocMemCache

If you want a server process-scoped, in-memory cache that lasts for the entire server process, we recommend you use this Django cache engine. You are free to use your own implementation, of course. You can also use this in combination with other cache engines (from this module or others) in a multi-cache Django configuration.

close(**_kwargs: Any) → None

There is no reason to ever clear the cache.

class pysoa.server.django.cache.PySOAPyLibMCCache(server, params)

Bases: django.core.cache.backends.memcached.PyLibMCCache

If you want to use Memcached with the pylibmc library, we recommend you use this Django cache engine, which does not close the socket connections to Memcached unless the server is shutting down and the super class supports closing the connections. You are free to use your own implementation, of course. You can also use this in combination with other cache engines (from this module or others) in a multi-cache Django configuration.

close(for_shutdown: bool = False, **_kwargs: Any) → None

Only call super().close() if the server is shutting down (not between requests).

Parameters

for_shutdown – If True, the socket connections to Memcached are closed (defaults to False).

class pysoa.server.django.cache.PySOARequestScopedMemoryCache(name, params)

Bases: django.core.cache.backends.locmem.LocMemCache

If you want a request-scoped, in-memory cache that clears at the end of each job request, we recommend you use this Django cache engine. You are free to use your own implementation, of course. You can also use this in combination with other cache engines (from this module or others) in a multi-cache Django configuration.

close(**_kwargs: Any) → None

Clear the cache completely at the end of each request.

pysoa.test.assertions.raises_call_action_error(**kwargs) → Iterator[pysoa.test.assertions._PySOAExceptionInfo]

pysoa.test.assertions.raises_error_codes(error_codes: Union[Iterable[str], str], only: bool = False, **kwargs: Any) → Iterator[pysoa.test.assertions._PySOAExceptionInfo]

pysoa.test.assertions.raises_field_errors(field_errors: Dict[str, Union[Iterable[str], str]], only: bool = False, **kwargs: Any) → Iterator[pysoa.test.assertions._PySOAExceptionInfo]

pysoa.test.assertions.raises_only_error_codes(error_codes: Union[Iterable[str], str], **kwargs: Any) → Iterator[pysoa.test.assertions._PySOAExceptionInfo]

pysoa.test.assertions.raises_only_field_errors(field_errors: Dict[str, Union[Iterable[str], str]], **kwargs: Any) → Iterator[pysoa.test.assertions._PySOAExceptionInfo]

class pysoa.test.server.BaseServerTestCase

Bases: object

Base class for all test classes that need to call the server. It runs calls to actions through the server stack so configured middleware runs and requests and responses go through the normal validation cycles. Note that this uses the local transports, so requests and responses are not serialized.

assertActionRunsWithAndReturnErrors(action: str, body: Dict[str, Any], **kwargs: Any) → List[pysoa.common.errors.Error]

Calls self.call_action and asserts that it runs with errors, and returns those errors.

assertActionRunsWithErrorCodes(action: str, body: Dict[str, Any], error_codes: Union[Iterable[str], str], only: bool = False, **kwargs: Any) → None

Calls self.call_action and asserts that it runs with the specified error codes.

Parameters

• action – The name of the action to call

• body – The request body to send to the action

• error_codes – A single error code or iterable of multiple error codes (all of the specified errors must be present in the response).

• only – If True additional errors cause a failure (defaults to False, so additional errors are ignored).

• kwargs – Additional keyword arguments to send to pysoa.client.client.Client.call_action().

assertActionRunsWithFieldErrors(action: str, body: Dict[str, Any], field_errors: Dict[str, Union[Iterable[str], str]], only: bool = False, **kwargs: Any) → None

Calls self.call_action and asserts that it runs with the specified field errors.

Parameters

• action – The name of the action to call

• body – The request body to send to the action

• field_errors – A dictionary of field name keys to error codes or iterables of error codes for the fields (all of the specified errors must be present in the response).

• only – If True additional errors cause a failure (defaults to False, so additional errors are ignored).

• kwargs – Additional keyword arguments to send to pysoa.client.client.Client.call_action().

assertActionRunsWithOnlyErrorCodes(action: str, body: Dict[str, Any], error_codes: Union[Iterable[str], str], **kwargs: Any) → None

Convenient alternative to calling assertActionRunsWithErrorCodes() that sets the only argument to True.

assertActionRunsWithOnlyFieldErrors(action: str, body: Dict[str, Any], field_errors: Dict[str, Union[Iterable[str], str]], **kwargs: Any) → None

Convenient alternative to calling assertActionRunsWithFieldErrors() that sets the only argument to True.

call_action(action: str, body: Dict[str, Any] = None, service_name: Optional[str] = None, **kwargs: Any) → <class 'pysoa.common.types.ActionResponse'>

A convenience method alternative to calling self.client.call_action that allows you to omit the service name.

Parameters

• action – The required action name to call

• body – The optional request body to send to the action

• service_name – The optional service name if you need to call a service other than the configured local testing service.

• kwargs – Additional keyword arguments to send to pysoa.client.client.Client.call_action().

Returns

The value returned from pysoa.client.client.Client.call_action().

server_class = None

The reference to your Server class, which must be set in order to use the service helpers in this class.

server_settings = None

A settings dict to use when instantiating your Server class. If not specified, the service helpers in this class will attempt to get settings from the configured Django or PySOA settings module.

setup_pysoa() → None

Sets up self.client for use in calling the local testing service. Requires you to configure server_class and server_settings class attributes.

class pysoa.test.server.PyTestServerTestCase

Bases: pysoa.test.server.BaseServerTestCase

An extension of BaseServerTestCase that calls BaseServerTestCase.setup_pysoa() in setup_method(). If you override setup_method in your test class, you must call super().setup_method(), or else it will not work properly.

This class will detect in your test class and call, if present, implementations of unittest-like setUpClass, tearDownClass, setUp, and tearDown from setup_class, teardown_class, setup_method, and teardown_method, respectively, and issue a deprecation warning in doing so. You should migrate to the standard PyTest form of these methods if you wish to use this class. This class also provides addCleanup(), which behaves the same as the same-named method in unittest and also issues a deprecation warning. All of these polyfills will be removed in PySOA 2.0.

This class also provides polyfills for unittest-like self.assert* and self.fail* methods. There is currently no plan to deprecate and remove this, but that may happen by PySOA 2.0, and you should endeavor to adopt standard assert-style assertions, as they provide better failure output in PyTest results.

addCleanup(function: Callable, *args: Any, **kwargs: Any) → None

Deprecated, to be removed in PySOA 2.0.

assertAlmostEqual(first: <class 'float'>, second: <class 'float'>, places: Optional[int] = None, msg: Optional[object] = None, delta: Optional[float] = None) → None

assertAlmostEquals(**kwargs)

assertCountEqual(first: Iterable, second: Iterable, msg: Optional[object] = None) → None

assertDictEqual(first: Dict[Any, Any], second: Dict[Any, Any], msg: Optional[object] = None) → None

assertEqual(first: Any, second: Any, msg: Optional[object] = None) → None

assertEquals(**kwargs)

assertFalse(expr: Any, msg: Optional[object] = None) → None

assertGreater(first: Any, second: Any, msg: Optional[object] = None) → None

assertGreaterEqual(first: Any, second: Any, msg: Optional[object] = None) → None

assertIn(member: Any, container: Union[Iterable[Any], Container[Any]], msg: Optional[object] = None) → None

assertIs(first: Any, second: Any, msg: Optional[object] = None) → None

assertIsInstance(obj: Any, msg: Optional[object] = None) → None

assertIsNone(expr: Any, msg: Optional[object] = None) → None

assertIsNot(first: Any, second: Any, msg: Optional[object] = None) → None

assertIsNotNone(expr: Any, msg: Optional[object] = None) → None

assertLess(first: Any, second: Any, msg: Optional[object] = None) → None

assertLessEqual(first: Any, second: Any, msg: Optional[object] = None) → None

assertListEqual(first: List[Any], second: List[Any], msg: Optional[object] = None) → None

assertLogs(logger: Union[str, bytes, logging.Logger, None] = None, level: Union[str, bytes, int, None] = None) → <class 'pysoa.test.server._AssertLogsContext'>

assertMultiLineEqual(first: str, second: str, msg: Optional[object] = None) → None

assertNotAlmostEqual(first: <class 'float'>, second: <class 'float'>, places: Optional[int] = None, msg: Optional[object] = None, delta: Optional[float] = None) → None

assertNotAlmostEquals(**kwargs)

assertNotEqual(first: Any, second: Any, msg: Optional[object] = None) → None

assertNotEquals(**kwargs)

assertNotIn(member: Any, container: Union[Iterable[Any], Container[Any]], msg: Optional[object] = None) → None

assertNotIsInstance(obj: Any, msg: Optional[object] = None) → None

assertNotRegex(text: ~_S, regex: Union[Pattern[~_S], ~_S], msg: Optional[object] = None) → None

assertNotRegexpMatches(**kwargs)

assertRaises(exception: Union[Type[BaseException], Tuple[Type[BaseException], ...]], callable: Callable = None, *args: Any, **kwargs: Any) → <class '_pytest.python_api.RaisesContext'>

assertRaisesRegex(exception: Union[Type[BaseException], Tuple[Type[BaseException], ...]], regex: Union[Pattern[~AnyStr], ~AnyStr], callable: Callable = None, *args: Any, **kwargs: Any) → <class '_pytest.python_api.RaisesContext'>

assertRaisesRegexp(**kwargs)

assertRegex(text: ~_S, regex: Union[Pattern[~_S], ~_S], msg: Optional[object] = None) → None

assertRegexpMatches(**kwargs)

assertSequenceEqual(first: Sequence[Any], second: Sequence[Any], msg: Optional[object] = None, seq_type: Union[Type[Sequence[Any]], None] = None) → None

assertSetEqual(first: AbstractSet[Any], second: AbstractSet[Any], msg: Optional[object] = None) → None

assertTrue(expr: Any, msg: Optional[object] = None) → None

assertTupleEqual(first: Tuple[Any, ...], second: Tuple[Any, ...], msg: Optional[object] = None) → None

assertWarns(exception: Union[Type[Warning], Tuple[Type[Warning], ...]], callable: Callable = None, *args: Any, **kwargs: Any) → <class '_pytest.recwarn.WarningsChecker'>

assertWarnsRegex(exception: Union[Type[Warning], Tuple[Type[Warning], ...]], regex: Union[Pattern[~AnyStr], ~AnyStr], callable: Callable = None, *args: Any, **kwargs: Any) → <class '_pytest.recwarn.WarningsChecker'>

assert_(**kwargs)

doCleanups() → bool

fail(msg: Optional[object] = None) → NoReturn

failIf(**kwargs)

failIfAlmostEqual(**kwargs)

failIfEqual(**kwargs)

failUnless(**kwargs)

failUnlessAlmostEqual(**kwargs)

failUnlessEqual(**kwargs)

failUnlessRaises(**kwargs)

setUp() → None

Deprecated, to be removed in PySOA 2.0. Override setup_method(), instead, and be sure to still call super().setup_method().

classmethod setUpClass() → None

Deprecated, to be removed in PySOA 2.0. Override setup_class(), instead, and be sure to still call super().setup_class().

classmethod setup_class() → None

setup_method() → None

tearDown() → None

Deprecated, to be removed in PySOA 2.0. Override teardown_method(), instead, and be sure to still call super().teardown_method().

classmethod tearDownClass() → None

Deprecated, to be removed in PySOA 2.0. Override teardown_class(), instead, and be sure to still call super().teardown_class().

classmethod teardown_class() → None

teardown_method() → None

pysoa.test.server.ServerTestCase

alias of pysoa.test.server.PyTestServerTestCase

class pysoa.test.server.UnitTestServerTestCase(methodName='runTest')

Bases: unittest.case.TestCase, pysoa.test.server.BaseServerTestCase

An extension of BaseServerTestCase that calls BaseServerTestCase.setup_pysoa() in setUp(). If you override setUp in your test class, you must call super().setUp(), or else it will not work properly.

setUp() → None

Hook method for setting up the test fixture before exercising it.

class pysoa.test.stub_service.stub_action(service, action, body=None, errors=None, side_effect=None, register_response_schema_contract=True, register_request_schema_contract=True)

Bases: object

Stub an action temporarily. This is useful for things like unit testing, where you really need to test the code calling a service, but you don’t want to test the actual service along with it, or the actual service isn’t easily available from the automated test process.

You can use this as a context manager or as a decorator, but you can only decorate classes, instance methods, and class methods. Decorating static methods and functions will cause it to barf on argument order.

Some example uses cases:

@stub_action('user', 'get_user', body={'user': {'id': 1234, 'username': 'John', 'email': 'john@example.org'}})
class TestSomeCode(object):
    '''
    This class is decorated to stub an action that the tested code ends up calling for all or most of these
    tests.
    '''

    def test_simple_user_helper(self, stub_get_user):
        user = UserHelper().get_user_from_service(user_id=5678)

        # This shows all the various assertions that can be made:
        #  - stub_get_user.called: a simple boolean, yes or no it was or wasn't called at least once
        #  - stub_get_user.call_count: the number of times it was called
        #  - stub_get_user.call_body: the request body for the last time the stub action was called
        #  - stub_get_user.call_bodies: a tuple of all request bodies for all times the action was called
        #  - stub_get_user extends MagicMock, so you can use things like ``assert_called_once_with`` and
        #    ``assert_has_calls`` like you would with any MagicMock (the value passed to the mock is always
        #    the request body), but many will find ``call_body`` and ``call_bodies`` easier to use.
        self.assertTrue(stub_get_user.called)
        self.assertEqual(1, stub_get_user.call_count)
        self.assertEqual({'id': 5678}, stub_get_user.call_body)
        self.assertEqual(({'id': 5678}, ), stub_get_user.call_bodies)
        stub_get_user.assert_called_once_with({'id': 5678})
        stub_get_user.assert_has_calls(
            mock.call({'id': 5678}),
        )

    @stub_action('settings', 'get_user_setting')
    def test_complex_user_helper(self, stub_get_user_setting, stub_get_user):
        # You can combine class and method decorators. As with ``mock.patch``, the order of the arguments is the
        # reverse of that which you would expect. You can combine class and/or function stub decorators with
        # ``mock.patch`` decorators, and the order of the various stubs and mocks will likewise follow the order
        # they are mixed together.

        # Instead of passing a body or errors to the stub decorator or context manager, you can add it to the
        # stub after creation (but before use). Since action stubs extend ``MagicMock``, you can use
        # ``return_value`` (it should be the response body dict) or ``side_effect`` (it should be ActionError(s) or
        # response body dict(s)). We use ``side_effect`` here to demonstrate expecting multiple calls.

        stub_get_user_setting.side_effect = (
            {'value': 'This is the first setting value response'},
            {'value': 'This is the second setting value response'},
            ActionError(errors=[Error(code='NO_SUCH_SETTING', message='The setting does not exist')]),
        )

        settings = UserHelper().get_user_settings(user_id=1234)

        self.assertEqual(
            {
                'setting1', 'This is the first setting value response',
                'setting2', 'This is the second setting value response',
            },
            settings,
        )

        self.assertEqual(3, stub_get_user_setting.call_count)
        self.assertEqual(
            (
                {'user_id': 1234, 'setting_id': 'setting1'},
                {'user_id': 1234, 'setting_id': 'setting2'},
                {'user_id': 1234, 'setting_id': 'setting3'}
            ),
            stub_get_user_setting.call_bodies,
        )

        stub_user.assert_called_once_with({'id': 1234})

    def test_another_user_helper_with_context_manager(self, stub_get_user):
        # Using a context manager is intuitive and works essentially the same as using a decorator

        with stub_action('payroll', 'get_salary') as stub_get_salary:
            stub_get_salary.return_value = {'salary': 75950}

            salary = UserHelper().get_user_salary(user_id=1234)

        self.assertEqual(75950, salary)

        self.assertEqual(1, stub_get_salary.call_count)
        self.assertEqual({'user_id': 1234}, stub_get_salary.call_body)

        stub_user.assert_called_once_with({'id': 1234})

__call__(func: ~_CT) → ~_CT

Call self as a function.

__enter__() → <class 'pysoa.test.stub_service.stub_action._MockAction'>

__exit__(exc_type: Any = None, exc_value: Any = None, traceback: Any = None) → typing_extensions.Literal[False]

__init__(service: str, action: str, body: Union[Dict[str, Any], None] = None, errors: Union[Iterable[Mapping[str, Any]], Iterable[pysoa.common.errors.Error], None] = None, side_effect: Union[Dict[str, Any], Exception, Type[Exception], Callable[[Dict[str, Any]], Dict[str, Any]], Iterable[Union[Dict[str, Any], Exception, Type[Exception], Callable[[Dict[str, Any]], Dict[str, Any]]]]] = None, register_response_schema_contract: bool = True, register_request_schema_contract: bool = True) → None

Initialize self. See help(type(self)) for accurate signature.

decorate_callable(func: ~_CT_C) → ~_CT_C

decorate_class(_class: ~_CT_T) → ~_CT_T

property is_local

start() → <class 'pysoa.test.stub_service.stub_action._MockAction'>

stop() → None

class pysoa.test.stub_service.StubClient(service_action_map=None, **_)

Bases: pysoa.client.client.Client

A Client for testing code that calls service actions.

Uses StubClientTransport, which incorporates a server for handling requests. Uses the real Server code path, so that developers needing to test their code against particular service responses can test against a genuine service in a unit testing environment.

__init__(service_action_map: Union[Mapping[str, Mapping[str, Mapping[str, Any]]], None] = None, **_: Any) → None

Generate settings based on a mapping of service names to actions.

Parameters

service_action_map – Dictionary of {service_name: {action_name: {'body': ..., 'errors': ...}}}

settings_class

alias of StubClientSettings

stub_action(service_name: str, action: str, body: Union[Dict[str, Any], None] = None, errors: Union[Iterable[Mapping[str, Any]], Iterable[pysoa.common.errors.Error], None] = None) → None

Stub the given action for the given service, configuring a handler and transport for that service if necessary.

Parameters

• service_name – The service name

• action – The action name

• body – The optional body to return

• errors – The optional errors to raise

class pysoa.test.stub_service.StubClientSettings(data)

Bases: pysoa.client.settings.ClientSettings

Settings Schema Definition

• metrics - dictionary with keys path and kwargs whose kwargs schema switches based on the value of path, dynamically based on class imported from path (see the configuration settings schema documentation for the class named at path). Configuration for defining a usage and performance metrics recorder. The imported item at the specified path must be a subclass of pymetrics.recorders.base.MetricsRecorder.

• middleware - list: The list of all ClientMiddleware objects that should be applied to requests made from this client to the associated service

  values

  dictionary with keys path and kwargs whose kwargs schema switches based on the value of path, dynamically based on class imported from path (see the configuration settings schema documentation for the class named at path). The imported item at the specified path must be a subclass of pysoa.client.middleware.ClientMiddleware.

• transport - dictionary with keys path and kwargs whose kwargs schema switches based on the value of path, dynamically based on class imported from path (see the configuration settings schema documentation for the class named at path). The imported item at the specified path must be a subclass of pysoa.common.transport.base.ClientTransport.

Default Values

Keys present in the dict below can be omitted from compliant settings dicts, in which case the values below will apply as the default values.

{
    "metrics": {
        "path": "pymetrics.recorders.noop:NonOperationalMetricsRecorder"
    },
    "middleware": [],
    "transport": {
        "path": "pysoa.test.stub_service:StubClientTransport"
    }
}

defaults = {'metrics': {'path': 'pymetrics.recorders.noop:NonOperationalMetricsRecorder'}, 'middleware': [], 'transport': {'path': 'pysoa.test.stub_service:StubClientTransport'}}

schema = {'metrics': ClassConfigurationSchema(base_class=<class 'pymetrics.recorders.base.MetricsRecorder'>, default_path=None, description='Configuration for defining a usage and performance metrics recorder.', eager_default_validation=True, add_class_object_to_dict=True), 'middleware': List(contents=ClassConfigurationSchema(base_class=<class 'pysoa.client.middleware.ClientMiddleware'>, default_path=None, description=None, eager_default_validation=True, add_class_object_to_dict=True), max_length=None, min_length=None, description='The list of all `ClientMiddleware` objects that should be applied to requests made from this client to the associated service', additional_validator=None), 'transport': ClassConfigurationSchema(base_class=<class 'pysoa.common.transport.base.ClientTransport'>, default_path=None, description=None, eager_default_validation=True, add_class_object_to_dict=True)}

class pysoa.test.stub_service.StubClientTransport(service_name='test', metrics=None, action_map=None)

Bases: pysoa.common.transport.local.LocalClientTransport

A transport that incorporates an automatically-configured StubServer for handling requests.

Class Configuration Schema

strict dict: The settings for the local transport

• action_map - flexible dict: (no description)

  keys

  unicode: The name of the action to stub

  values

  strict dict: A dictionary containing either a body dict or an errors list, providing an instruction on how the stub action should respond to requests

  • body - flexible dict: The body with which the action should respond, if no errors

    keys

    hashable: (no description)

    values

    anything: (no description)

  • errors - list: The errors with which the action should respond, if no body

    values

    any of the types bulleted below: (no description)

    • a Python object that is an instance of the following class or classes: pysoa.common.errors.Error.

    • strict dict: (no description)

      • code - unicode: (no description)

      • denied_permissions - list: (no description)

        values

        unicode: (no description)

      • field - unicode: (no description)

      • message - unicode: (no description)

      • traceback - unicode: (no description)

      • variables - flexible dict: (no description)

        keys

        hashable: (no description)

        values

        anything: (no description)

      Optional keys: denied_permissions, field, traceback, variables

  Optional keys: body, errors

Optional keys: action_map

__init__(service_name: str = 'test', metrics: Optional[pymetrics.recorders.base.MetricsRecorder] = None, action_map: Union[Mapping[str, Mapping[str, Any]], None] = None) → None

Configure a StubServer to handle requests. Creates a new subclass of StubServer using the service name and action mapping provided.

Parameters

• service_name – The service name.

• metrics – You can omit this, but if you really want, override the default NoOpMetricsRecorder.

• action_map – Dictionary of {action_name: {'body': action_body, 'errors': action_errors}} where action_body is a dictionary and action_errors is a list.

stub_action(action: str, body: Union[Dict[str, Any], None] = None, errors: Union[Iterable[Mapping[str, Any]], Iterable[pysoa.common.errors.Error], None] = None) → None

Stub the given action with the configured server.

Parameters

• action – The action name

• body – The optional body to return

• errors – The optional errors to raise

class pysoa.test.stub_service.StubClientTransportSchema(contents=None, optional_keys=frozenset({}), allow_extra_keys=None, description=None, additional_validator=None)

Bases: conformity.fields.structures.Dictionary

contents = {'action_map': SchemalessDictionary(key_type=UnicodeString(min_length=None, max_length=None, description='The name of the action to stub', allow_blank=True), value_type=Dictionary(contents={'body': SchemalessDictionary(key_type=Hashable(description=None), value_type=Anything(description=None), max_length=None, min_length=None, description='The body with which the action should respond, if no errors', additional_validator=None), 'errors': List(contents=Any(), max_length=None, min_length=None, description='The errors with which the action should respond, if no body', additional_validator=None)}, optional_keys=frozenset({'body', 'errors'}), allow_extra_keys=False, description='A dictionary containing either a body dict or an errors list, providing an instruction on how the stub action should respond to requests', additional_validator=None), max_length=None, min_length=None, description=None, additional_validator=None)}

description = 'The settings for the local transport'

optional_keys = ('action_map',)

class pysoa.test.stub_service.StubServer(settings, forked_process_id=None)

Bases: pysoa.server.server.Server

A Server that provides an interface to stub actions, i.e. define actions inline, for testing purposes.

stub_action(action, body=None, errors=None)

Make a new StubAction class with the given body and errors, and add it to the action_class_map.

The name of the action class is the action name converted to camel case. For example, an action called ‘update_foo’ will have an action class called UpdateFoo.

Parameters

• action – The action name

• body – The optional body to return

• errors – The optional errors to raise

Copyright © 2021 Eventbrite, freely licensed under Apache License, Version 2.0.

Documentation generated 2021 September 15 20:46 UTC.