Skip to main content

luna.kernel.handle

KernelHandle Objects​

class KernelHandle(ABC)

Abstract base class defining the core functionalities for interacting with the Luna kernel. It specifies methods to add contracts, tick contracts, fetch template and contract information, validate parameters, and handle events. This class is intended to be subclassed to provide concrete implementations for local and external kernel interactions.

add_contract​

@abstractmethod
def add_contract(
        contract_id: str, parameters: dict[str,
                                           JsonValue], roles: dict[str,
                                                                   list[str]],
        linked_contracts: dict[str, str]) -> types.AddContractResponse

Abstract method for adding a contract to the kernel.

tick​

@abstractmethod
def tick(contract_id: str, tick_id: str,
         contract_clock_offset: timedelta) -> types.TickResponse

Abstract method for ticking a contract. Ticking a contract involves advancing its state based on the provided clock offset.

template_info​

@abstractmethod
def template_info(template_definition: str,
                  template_class_name: str) -> types.TemplateInfoResponse

Abstract method for retrieving information about a template.

project_info​

@abstractmethod
def project_info(compressed_repository: bytes) -> types.ProjectInfoResponse

Abstract method for retrieving information about a project repository.

contract_info​

@abstractmethod
def contract_info(contract_id: str) -> types.ContractInfoResponse

Abstract method for fetching information about a specific contract.

validate_parameters​

@abstractmethod
def validate_parameters(
    template_definition: str,
    parameters: dict[str, JsonValue],
    contract_id: str,
    template_class_name: str | None = None
) -> types.ValidateParametersResponse

Abstract method for validating the parameters of a contract or template.

put_event​

@abstractmethod
def put_event(contract_id: str,
              event_name: str,
              source_type: EventSourceType,
              source_id: str,
              payload: dict[str, JsonValue],
              publish_at: datetime | None = None) -> types.PutEventResponse

Abstract method for posting an event to a contract.

KernelHandleFactory Objects​

class KernelHandleFactory(ABC, Generic[KER])

Class: KernelHandleFactory​

Abstract base class that has a single function which returns an initialized KernelHandle. Used to initialize KernelHandles at runtime.

LocalKernelHandle Objects​

class LocalKernelHandle(KernelHandle)

Concrete implementation of the KernelHandle for local kernel interactions. This class directly interfaces with a local instance of the Luna kernel API.

add_contract​

def add_contract(
        contract_id: str, parameters: dict[str,
                                           JsonValue], roles: dict[str,
                                                                   list[str]],
        linked_contracts: dict[str, str]) -> types.AddContractResponse

Adds a contract to the local kernel. The method constructs an AddContractRequest and sends it to the local kernel API, returning the response.

tick​

def tick(contract_id: str, tick_id: str,
         contract_clock_offset: timedelta) -> types.TickResponse

Ticks a contract in the local kernel. The method constructs a TickRequest with the contract's ID, tick ID, and clock offset, then sends it to the local kernel API.

template_info​

def template_info(template_definition: str,
                  template_class_name: str) -> types.TemplateInfoResponse

Retrieves information about a template from the local kernel. The method constructs a TemplateInfoRequest and sends it to the local kernel API.

contract_info​

def contract_info(contract_id: str) -> types.ContractInfoResponse

Fetches information about a specific contract from the local kernel. The method constructs a ContractInfoRequest and sends it to the local kernel API.

validate_parameters​

def validate_parameters(
    template_definition: str,
    parameters: dict[str, JsonValue],
    contract_id: str,
    template_class_name: str | None = None
) -> types.ValidateParametersResponse

Validates the parameters of a contract or template against the local kernel. The method constructs a ValidateParametersRequest and sends it to the local kernel API.

put_event​

def put_event(contract_id: str,
              event_name: str,
              source_type: EventSourceType,
              source_id: str,
              payload: dict[str, JsonValue],
              publish_at: datetime | None = None) -> types.PutEventResponse

Posts an event to a contract in the local kernel. The method constructs a PutEventRequest and sends it to the local kernel API.

ExternalKernelHandle Objects​

class ExternalKernelHandle(KernelHandle)

Concrete implementation of the KernelHandle for external kernel interactions. This class communicates with an external kernel service via HTTP requests.

healthcheck​

def healthcheck()

Performs a health check on the external kernel service.

make_request​

def make_request(path: str, data: LunaBaseModel,
                 response_ser: Type[SER]) -> SER

Helper method to make an HTTP POST request to the external kernel service.

add_contract​

def add_contract(
        contract_id: str, parameters: dict[str,
                                           JsonValue], roles: dict[str,
                                                                   list[str]],
        linked_contracts: dict[str, str]) -> types.AddContractResponse

Adds a contract to the external kernel. The method constructs an AddContractRequest payload and sends it via the make_request method.

tick​

def tick(contract_id: str, tick_id: str,
         contract_clock_offset: timedelta) -> types.TickResponse

Ticks a contract in the external kernel. The method constructs a TickRequest payload and sends it via the make_request method.

template_info​

def template_info(template_definition: str,
                  template_class_name: str) -> types.TemplateInfoResponse

Retrieves information about a template from the external kernel. The method constructs a TemplateInfoRequest payload and sends it via the make_request method.

contract_info​

def contract_info(contract_id: str) -> types.ContractInfoResponse

Fetches information about a specific contract from the external kernel. The method constructs a ContractInfoRequest payload and sends it via the make_request method.

validate_parameters​

def validate_parameters(
    template_definition: str,
    parameters: dict[str, JsonValue],
    contract_id: str,
    template_class_name: str | None = None
) -> types.ValidateParametersResponse

Validates the parameters of a contract or template against the external kernel. The method constructs a ValidateParametersRequest payload and sends it via the make_request method.

put_event​

def put_event(contract_id: str,
              event_name: str,
              source_type: EventSourceType,
              source_id: str,
              payload: dict[str, JsonValue],
              publish_at: datetime | None = None) -> types.PutEventResponse

Posts an event to a contract in the external kernel. The method constructs a PutEventRequest payload and sends it via the make_request method.

LocalKernelHandleFactory Objects​

class LocalKernelHandleFactory(KernelHandleFactory[LocalKernelHandle])

LocalKernelHandleFactory returns a factory that creates Kernel Handles which use a local version of the kernel and engine.

RemoteKernelHandleFactory Objects​

class RemoteKernelHandleFactory(KernelHandleFactory[LocalKernelHandle])

RemoteKernelHandleFactory returns a factory that creates Kernel Handles which use a local version of the kernel but a remote version of the engine

ExternalKernelHandleFactory Objects​

class ExternalKernelHandleFactory(KernelHandleFactory[ExternalKernelHandle])

ExternalKernelHandleFactory returns a factory that creates Kernel Handles which use both an external version of the kernel and engine