Service Definitions

Core decorators (and their associated types) for INTERSECT. Only relevant for Service authors.

Users annotate their endpoints with these decorators (chiefly @intersect_message()) much as they would when developing a traditional REST API. The decorator allows for users to declare certain key aspects about their data, without having to delve into the plumbing which integrates their data into other backing services.

When annotating a function with @intersect_message() or @intersect_status(), you must make sure all function parameters and the function return value are annotated with valid types. This is necessary for generating a schema when creating an IntersectService. If you are not able to create a schema, the service will refuse to start.

final class intersect_sdk.service_definitions.IntersectEventDefinition(*, event_type: Any, event_documentation: str = '', content_type: Annotated[str, _PydanticGeneralMetadata(pattern='\\w+/[-+.\\w]+')] = 'application/json', data_handler: IntersectDataHandler = IntersectDataHandler.MESSAGE)

When defining your dictionary/map of events, the values will be represented by an EventDefinition.

content_type: Annotated[str, FieldInfo(annotation=NoneType, required=True, metadata=[_PydanticGeneralMetadata(pattern='\\w+/[-+.\\w]+')])]

The IntersectMimeType (aka Content-Type) of your event.

default: ‘application/json’

data_handler: IntersectDataHandler

The IntersectDataHandler you want to use (most people can just use IntersectDataHandler.MESSAGE here, unless your data is very large)

default: IntersectDataHandler.MESSAGE

event_documentation: str

This is a strictly informational field which can describe what the event does in schema.

event_type: Any

The data TYPE of your event. Note that any event you emit must map to this type.

The type you provide must be parsable by Pydantic.

intersect_sdk.service_definitions.intersect_message(__func: Callable[[...], Any] | None = None, /, *, ignore_keys: set[str] | None = None, request_content_type: Annotated[str, FieldInfo(annotation=NoneType, required=True, metadata=[_PydanticGeneralMetadata(pattern='\\w+/[-+.\\w]+')])] = 'application/json', response_data_transfer_handler: IntersectDataHandler = IntersectDataHandler.MESSAGE, response_content_type: Annotated[str, FieldInfo(annotation=NoneType, required=True, metadata=[_PydanticGeneralMetadata(pattern='\\w+/[-+.\\w]+')])] = 'application/json', strict_request_validation: bool = False) → Callable[[...], Any]

Use this annotation to mark your capability method as an entrypoint to external requests.

A class method marked with this annotation has two requirements:

  1. Excluding the self-reference in the class definition, any additional parameters MUST have an entire class hierarchy of the following types (for a comprehensive overview of types, see https://docs.pydantic.dev/latest/concepts/types/):

  • Pydantic’s “BaseModel” class.

  • Dataclasses (You can either use Pydantic’s or the standard library’s)

  • TypedDict

  • NamedTuple

  • primitive types (str, bytes, int, float, bool)

  • None

  • Union/Optional types

  • Iterable/Sequence types (list, deque, set, tuple, frozenset, etc.)

  • Mapping types (dict, Counter, OrderedDict, etc.). Regarding mapping types: the keys must be one of str/float/int, and float/int keys CANNOT use strict_request_validation=True.

  • most stdlib types, i.e. Decimal, datetime.datetime, pathlib, etc.

  • using the typing.Annotated type in conjunction with Pydantic’s “Field” or various classes from the annotated_types library

  • TODO: Generators are a WORK IN PROGRESS but will eventually represent a streaming function

You are only allowed to have one additional parameter. Functions without this parameter are assumed to take in no arguments. Be sure to specify the parameter type in your function signature!

  1. Your response type MUST have a class hierarchy of the same types as above. Be sure to specify the parameter type in your function signature! (If you do not return a response, explicitly mark your return type as “None”)

Example:

@intersect_message()
def some_external_function(self, request: MyBaseModelRequest) -> MyBaseModelResponse:
    # be sure to return "MyBaseModelResponse" here

In general, if you are able to create a service from this class, you should be okay.

Params:

  • ignore_keys: Hashset of keys. The service class maintains a set of keys to ignore, and will ignore this function if at least one key is present in the service set. By default, all functions will always be allowed. You generally only need this if you want to forbid only a subset of functions - use “service.shutdown()” to disconnect from INTERSECT entirely. In general, you should NOT define this on functions which are just query functions; only set this if you are mutating INSTRUMENT or APPLICATION state.

  • request_content_type: how to deserialize incoming requests (default: application/json)

  • response_content_type: how to serialize outgoing requests (default: application/json)

  • response_data_transfer_handler: are responses going out through the message, or through another mean (i.e. MINIO)?

  • strict_request_validation: if this is set to True, use pydantic strict validation for requests - otherwise, use lenient validation (default: False) See https://docs.pydantic.dev/latest/concepts/conversion_table/ for more info about this. NOTE: If you are using a Mapping type (i.e. Dict) with integer or float keys, you MUST leave this on False.

intersect_sdk.service_definitions.intersect_status(__func: Callable[[...], Any] | None = None, /) → Any

Use this annotation to mark your capability method as a status retrieval function.

You may ONLY mark ONE function as a status retrieval function. It’s advisable to have one.

Your status retrieval function may not have any parameters (other than “self”). Return annotation rules mirror the typing rules for @intersect_message().

A status retrieval function should ALWAYS return valid JSON. You should not be returning large globs of data, a few KB serialized should be sufficient.

A status message MUST NOT send events out. It should be a simple query of the general service (no specifics). A status message MUST send its response back in a value which can be serialized into JSON. A status message MUST have a fairly small response size (no large data).