Skip to content

runtime.middleware

DefaultMiddlewareNamespace

Source code in /home/runner/work/arc/arc/arc/runtime/middleware.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
class DefaultMiddlewareNamespace:
    _list: list[Middleware]

    @classmethod
    def all(cls) -> list[Middleware]:
        """Returns a list of all default middlewares"""
        return cls._list

    @classmethod
    def regsiter(cls, name: str, middleware: Middleware) -> None:
        """Register a new default middleware."""
        cls._list.append(middleware)
        setattr(cls, name, middleware)

all() classmethod

Returns a list of all default middlewares

Source code in /home/runner/work/arc/arc/arc/runtime/middleware.py 
244
245
246
247
@classmethod
def all(cls) -> list[Middleware]:
    """Returns a list of all default middlewares"""
    return cls._list

regsiter(name, middleware) classmethod

Register a new default middleware.

Source code in /home/runner/work/arc/arc/arc/runtime/middleware.py 
249
250
251
252
253
@classmethod
def regsiter(cls, name: str, middleware: Middleware) -> None:
    """Register a new default middleware."""
    cls._list.append(middleware)
    setattr(cls, name, middleware)

MiddlewareManager

Source code in /home/runner/work/arc/arc/arc/runtime/middleware.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
class MiddlewareManager:
    def __init__(self, middlewares: t.Sequence[Middleware]):
        self._stack = MiddlewareStack(middlewares)

    @t.overload
    def use(
        self,
        handler: None = None,
        *,
        pos: int | None = None,
        replace: Middleware | None = None,
        before: Middleware | None = None,
        after: Middleware | None = None,
    ) -> t.Callable[[Middleware], Middleware]:
        ...

    @t.overload
    def use(
        self,
        handler: Middleware,
        *,
        pos: int | None = None,
        replace: Middleware | None = None,
        before: Middleware | None = None,
        after: Middleware | None = None,
    ) -> Middleware:
        ...

    @t.overload
    def use(
        self,
        handler: t.Sequence[Middleware],
        *,
        pos: int | None = None,
        replace: Middleware | None = None,
        before: Middleware | None = None,
        after: Middleware | None = None,
    ) -> t.Sequence[Middleware]:
        ...

    def use(
        self,
        handler: t.Any = None,
        *,
        pos: int | None = None,
        replace: Middleware | None = None,
        before: Middleware | None = None,
        after: Middleware | None = None,
    ) -> t.Any:
        """Register a middleware with this object.

        Args:
            handler (callable, optional): Callable to register as a middleware.
                Can optionally receive an array of middlewares.
            pos (int | None, optional): Index in the middleware stack to insert this middleware. Defaults to None.
            replace (Middleware | None, optional): A middleware you want to replace with this middleware. Defaults to None.
            before (Middleware | None, optional): Insert the provided `handler` before this middleware. Defaults to None.
            after (Middleware | None, optional): Insert the provided `handler` after this middleware. Defaults to None.
        """

        def ensure_single_operation() -> None:
            ops = [op for op in (pos, replace, before, after) if op is not None]
            if len(ops) > 1:
                raise errors.InternalError(
                    "Cannot provide multiple operations for a single middleware"
                )

        def inner(handler: Middleware) -> Middleware:
            ensure_single_operation()
            if pos is not None:
                self._stack.insert(pos, handler)
            elif replace:
                idx = self._stack.index(replace)
                self._stack[idx] = handler
            elif before:
                idx = self._stack.index(before)
                self._stack.insert(idx - 1, handler)
            elif after:
                idx = self._stack.index(after)
                self._stack.insert(idx + 1, handler)
            else:
                self._stack.append(handler)

            return handler

        if handler:
            if callable(handler):
                return inner(handler)
            else:
                return [inner(f) for f in handler]

        return inner

    @t.overload
    def handle(  # type: ignore
        self, *exceptions: type[BaseException]
    ) -> t.Callable[[ErrorHandler], ErrorHandler]:
        ...

    @t.overload
    def handle(
        self, handler: ErrorHandler, *exceptions: type[BaseException]
    ) -> ErrorHandler:
        ...

    @t.overload
    def handle(
        self, handler: t.Sequence[ErrorHandler], *exceptions: type[BaseException]
    ) -> list[ErrorHandler]:
        ...

    def handle(self, handler: t.Any | type = None, *exceptions: t.Any) -> t.Any:
        """Register an exception handler to this object

        Args:
            handler (ErrorHandler, optional): Error handler callback, receives the context
                and the exception object.
        """

        def inner(func: ErrorHandler) -> ErrorHandler:
            @self.use
            def error_handler(ctx: Context) -> t.Any:
                try:
                    yield
                except exceptions as e:
                    return func(ctx, e)

            return func

        if issubclass(handler, BaseException):
            exceptions = (handler, *exceptions)
            return inner
        elif callable(handler):
            return inner(handler)
        else:
            return [inner(f) for f in handler]

handle(handler=None, *exceptions)

Register an exception handler to this object

Parameters:

Name Type Description Default
handler ErrorHandler

Error handler callback, receives the context and the exception object.

 None
Source code in /home/runner/work/arc/arc/arc/runtime/middleware.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
def handle(self, handler: t.Any | type = None, *exceptions: t.Any) -> t.Any:
    """Register an exception handler to this object

    Args:
        handler (ErrorHandler, optional): Error handler callback, receives the context
            and the exception object.
    """

    def inner(func: ErrorHandler) -> ErrorHandler:
        @self.use
        def error_handler(ctx: Context) -> t.Any:
            try:
                yield
            except exceptions as e:
                return func(ctx, e)

        return func

    if issubclass(handler, BaseException):
        exceptions = (handler, *exceptions)
        return inner
    elif callable(handler):
        return inner(handler)
    else:
        return [inner(f) for f in handler]

use(handler=None, *, pos=None, replace=None, before=None, after=None)

Register a middleware with this object.

Parameters:

Name Type Description Default
handler callable

Callable to register as a middleware. Can optionally receive an array of middlewares.

 None
pos int | None

Index in the middleware stack to insert this middleware. Defaults to None.

 None
replace Middleware | None

A middleware you want to replace with this middleware. Defaults to None.

 None
before Middleware | None

Insert the provided handler before this middleware. Defaults to None.

 None
after Middleware | None

Insert the provided handler after this middleware. Defaults to None.

 None
Source code in /home/runner/work/arc/arc/arc/runtime/middleware.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
def use(
    self,
    handler: t.Any = None,
    *,
    pos: int | None = None,
    replace: Middleware | None = None,
    before: Middleware | None = None,
    after: Middleware | None = None,
) -> t.Any:
    """Register a middleware with this object.

    Args:
        handler (callable, optional): Callable to register as a middleware.
            Can optionally receive an array of middlewares.
        pos (int | None, optional): Index in the middleware stack to insert this middleware. Defaults to None.
        replace (Middleware | None, optional): A middleware you want to replace with this middleware. Defaults to None.
        before (Middleware | None, optional): Insert the provided `handler` before this middleware. Defaults to None.
        after (Middleware | None, optional): Insert the provided `handler` after this middleware. Defaults to None.
    """

    def ensure_single_operation() -> None:
        ops = [op for op in (pos, replace, before, after) if op is not None]
        if len(ops) > 1:
            raise errors.InternalError(
                "Cannot provide multiple operations for a single middleware"
            )

    def inner(handler: Middleware) -> Middleware:
        ensure_single_operation()
        if pos is not None:
            self._stack.insert(pos, handler)
        elif replace:
            idx = self._stack.index(replace)
            self._stack[idx] = handler
        elif before:
            idx = self._stack.index(before)
            self._stack.insert(idx - 1, handler)
        elif after:
            idx = self._stack.index(after)
            self._stack.insert(idx + 1, handler)
        else:
            self._stack.append(handler)

        return handler

    if handler:
        if callable(handler):
            return inner(handler)
        else:
            return [inner(f) for f in handler]

    return inner

MiddlewareStack

Bases: collections.UserList[Middleware]

Source code in /home/runner/work/arc/arc/arc/runtime/middleware.py 
 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
class MiddlewareStack(collections.UserList[Middleware]):
    __gens: list[t.Generator[None, t.Any, None]]

    def __repr__(self) -> str:
        return f"MiddlewareStack({self.data!r})"

    def start(self, ctx: Context) -> Context:
        self.__gens = []

        for handler in self:
            res = handler(ctx)
            if isinstance(res, types.GeneratorType):
                self.__gens.append(res)
                res = next(res)

            res = t.cast(t.Union["Context", None], res)

            if res is not None:
                ctx = res

        return ctx

    def close(self, result: t.Any) -> t.Any:
        """Closes each callback by calling `next()` on them"""
        for gen in reversed(self.__gens):
            try:
                gen.send(result)
            except StopIteration as e:
                if e.value is not None:
                    result = e.value

        return result

    def throw(self, exception: Exception) -> None:
        """Used if an error occurs in command execution.
        Notifies each of the callbacks that an error occured.

        Args:
            exception: The exception that occured within the executing command

        Raises:
            exception: if none of the callbacks handle the exception, re-raises
        """

        exc_type = type(exception)
        trace = exception.__traceback__

        exception_handled = False

        for gen in reversed(self.__gens):
            try:
                if exception_handled:
                    try:
                        next(gen)
                    except StopIteration:
                        ...
                else:
                    gen.throw(exc_type, exception, trace)
            except StopIteration:
                exception_handled = True
            except Exception as e:
                exception = e
                exc_type = type(e)
                trace = e.__traceback__

        if not exception_handled:
            raise exception

    def try_remove(self, m: Middleware) -> None:
        try:
            self.remove(m)
        except ValueError:
            ...

close(result)

Closes each callback by calling next() on them

Source code in /home/runner/work/arc/arc/arc/runtime/middleware.py 
50
51
52
53
54
55
56
57
58
59
def close(self, result: t.Any) -> t.Any:
    """Closes each callback by calling `next()` on them"""
    for gen in reversed(self.__gens):
        try:
            gen.send(result)
        except StopIteration as e:
            if e.value is not None:
                result = e.value

    return result

throw(exception)

Used if an error occurs in command execution. Notifies each of the callbacks that an error occured.

Parameters:

Name Type Description Default
exception Exception

The exception that occured within the executing command

 required

Raises:

Type Description
exception

if none of the callbacks handle the exception, re-raises
Source code in /home/runner/work/arc/arc/arc/runtime/middleware.py 
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
def throw(self, exception: Exception) -> None:
    """Used if an error occurs in command execution.
    Notifies each of the callbacks that an error occured.

    Args:
        exception: The exception that occured within the executing command

    Raises:
        exception: if none of the callbacks handle the exception, re-raises
    """

    exc_type = type(exception)
    trace = exception.__traceback__

    exception_handled = False

    for gen in reversed(self.__gens):
        try:
            if exception_handled:
                try:
                    next(gen)
                except StopIteration:
                    ...
            else:
                gen.throw(exc_type, exception, trace)
        except StopIteration:
            exception_handled = True
        except Exception as e:
            exception = e
            exc_type = type(e)
            trace = e.__traceback__

    if not exception_handled:
        raise exception