harmos-packaging 0.1.5

Harmos packaging: reusable Cargo builds, inert inspection and guest componentization.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104

# ***
#
# Harmos
#
# SPDX-License-Identifier: Apache-2.0
#
# ***

"""Typed Operation Descriptors And Structural Transport Boundaries"""

# === Imports === #

# Standard Library Imports
from collections.abc import AsyncIterator
from dataclasses import dataclass
from typing import Literal, Protocol

# Third Party Imports
from google.protobuf.message import Message

# Local Imports

# Type Checking Imports

# === Public Exports === #

# === Constants === #

# === Types === #

Mode = Literal["unary", "client_stream", "server_stream", "bidi"]

# === Classes === #


@dataclass(frozen=True, slots=True)
class Operation[Request: Message, Response: Message, Input: Message, Output: Message]:
    id: str
    version: int
    mode: Mode
    codec: Literal["protobuf"]
    request: type[Request]
    response: type[Response]
    input: type[Input] | None
    output: type[Output] | None
    grpc_path: str
    rpc_request: type[Message]
    rpc_response: type[Message]
    contract: str
    fingerprint: str


class OperationError(RuntimeError):
    """A Structured Refusal Reported By An Invoker"""

    def __init__(self, code: int, message: str) -> None:
        super().__init__(message)
        self.code = code
        self.message = message


class ResponseStream[Response: Message, Output: Message](Protocol):
    """Output Items And The Operation's Declared Final Response"""

    def __aiter__(self) -> AsyncIterator[Output]: ...

    async def response(self) -> Response: ...


class Invoker(Protocol):
    """Transport Implementations Own Connections, Limits And Cancellation"""

    async def call[Request: Message, Response: Message](
        self,
        operation: Operation[Request, Response, Message, Message],
        request: Request,
    ) -> Response: ...

    async def upload[Request: Message, Response: Message, Input: Message](
        self,
        operation: Operation[Request, Response, Input, Message],
        request: Request,
        items: AsyncIterator[Input],
    ) -> Response: ...

    def download[Request: Message, Response: Message, Output: Message](
        self,
        operation: Operation[Request, Response, Message, Output],
        request: Request,
    ) -> ResponseStream[Response, Output]: ...

    def exchange[Request: Message, Response: Message, Input: Message, Output: Message](
        self,
        operation: Operation[Request, Response, Input, Output],
        request: Request,
        items: AsyncIterator[Input],
    ) -> ResponseStream[Response, Output]: ...


# === Public Functions === #

# === Private Functions === #

# === Entry Point === #