Skip to content

Wrapper

wrapper

EnvironmentVars

Bases: TypedDict

Required keys for CascadeWrapperBase's environmentVariables argument.

Source code in cascade_cms/wrapper.py 
15
16
17
18
19
20
class EnvironmentVars(TypedDict):
    """Required keys for `CascadeWrapperBase`'s `environmentVariables` argument."""

    SERVER: str
    API_KEY: str
    CASCADE_URL: str

CascadeWrapperBase

Context-manager entry point tying together the logger, REST driver, and Operations builder for a single script/session.

Use as

with CascadeWrapperBase(env_vars, config_vars) as cascade: cascade.operations.read(identifier) results = cascade.submit_requests()

Source code in cascade_cms/wrapper.py 
 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
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
class CascadeWrapperBase:
    """Context-manager entry point tying together the logger, REST driver,
    and Operations builder for a single script/session.

    Use as:
        with CascadeWrapperBase(env_vars, config_vars) as cascade:
            cascade.operations.read(identifier)
            results = cascade.submit_requests()
    """

    def __enter__(self):
        return self

    def __init__(
        self,
        environmentVariables: EnvironmentVars,
        configurationVariables: dict[str, Any] | None,
        debug: dict[str, Any] | None = None,
    ):
        """Initialize the logger, driver, and operations builder.

        Args:
            environmentVariables: Must contain "SERVER" (label used in log
                output), "API_KEY" (Cascade bearer token), and
                "CASCADE_URL" (base URL of the Cascade instance).
            configurationVariables: kwargs forwarded to the driver's cache
                backend (`SQLiteBackend`); pass an empty dict for defaults.
            debug: Optional debug config for `OperationLogger` (verbose
                nested logging); None enables normal/minimal logging.
        """
        self._logger = OperationLogger(
            server=environmentVariables["SERVER"],
            debug_config=debug,
        )
        self._driver = CascadeCMSRestDriver(
            environmentVariables['API_KEY'],
            environmentVariables['CASCADE_URL'],
            configurationVariables,
            logger=self._logger,
        )
        self.operations = Operations(self._driver, _logger=self._logger)

        self._logger.log_init(
            environmentVariables['CASCADE_URL'],
            os.path.basename(sys.argv[0]),
        )

    def __exit__(self, exc_type, exc_value, traceback):
        """Log session exit and close the driver, then propagate exceptions.

        Any exception raised inside the `with` block (other than
        `RuntimeWarning`) is re-raised after cleanup runs.
        """
        try:
            self._logger.log_exit()
            self._driver.close()
        except Exception as e:  # noqa: BLE001 - log cleanup failure without masking the original exception
            self._logger.log_python_error(e)

        if exc_type is not None and not isinstance(exc_type, RuntimeWarning):
            return False  # Propagate the exception

    @overload
    def submit_requests(self, result_type: type[T], *, executor: Executor | None = None) -> list[T]: ...
    @overload
    def submit_requests(self, *, executor: Executor | None = None) -> list[CascadeObjects]: ...

    def submit_requests(self, result_type: type[T] | None = None, *, executor: Executor | None = None) -> list[CascadeObjects] | list[T]:
        """
        Run every registered operation chain and return one result per chain.

        Chains run concurrently, but the nodes inside a chain run strictly in
        order — each operation or callback receives the previous node's
        result. A chain stops at its first failure without affecting any
        other chain.

        Results line up with the chains **in the order they were created**,
        so `results[0]` belongs to the first chain built. Failures appear in
        that list rather than being dropped: a `CascadeError` for an API
        failure, or the exception object a callback raised. They are
        returned as values, not raised — a missed `isinstance(result,
        CascadeError | Exception)` check lets a failure flow downstream as
        if it were a normal result.

        The chain list is cleared afterwards, so callbacks registered for one
        batch never run again in the next.

        Args:
            result_type: Type hint for Pylance/mypy (e.g., submit_requests(Asset))
            executor: Optional Executor for sync callbacks.
                     Use ThreadPoolExecutor (default) for I/O-bound work.
                     Use ProcessPoolExecutor(max_workers=<cpu_count>) for CPU-bound work.

        Example (CPU-bound callbacks):
            from concurrent.futures import ProcessPoolExecutor
            from os import cpu_count

            with ProcessPoolExecutor(max_workers=cpu_count()) as executor:
                cascade.operations.read(id).then(optimize_image)
                results = cascade.submit_requests(executor=executor)

        Returns:
            One entry per chain: its final result, or the error that stopped it.
        """
        chains = list(self.operations._chains)
        if not chains:
            return []

        self._logger.log_batch_start()
        try:
            results = self._driver.eventLoop.run_until_complete(
                self._execute_chains(chains, executor)
            )
        except Exception as e:  # noqa: BLE001 - top-level entry point must not raise; log and return empty
            self._logger.log_python_error(e)
            return []
        finally:
            self.operations._reset_chains()

        succeeded = sum(1 for r in results if not isinstance(r, CascadeError | Exception))
        self._logger.log_batch_end(succeeded, len(chains))
        return results

    async def _execute_chains(
        self,
        chains: list[OperationChain],
        executor: Executor | None = None,
    ) -> list[Any]:
        """
        Run every chain concurrently and collect their results in chain order.

        `return_exceptions=True` is a backstop only: a chain already reports
        operation and callback failures as values, so an exception here means
        the chain machinery itself broke, and one broken chain must not take
        the rest of the batch down with it.

        Args:
            chains: The chains to run.
            executor: Optional Executor for sync callbacks.

        Returns:
            One entry per chain, in the order the chains were created.
        """
        results = await asyncio.gather(
            *(chain.execute_async(executor) for chain in chains),
            return_exceptions=True,
        )
        return list(results)

__init__(environmentVariables, configurationVariables, debug=None)

Initialize the logger, driver, and operations builder.

Parameters:

Name Type Description Default
environmentVariables EnvironmentVars

Must contain "SERVER" (label used in log output), "API_KEY" (Cascade bearer token), and "CASCADE_URL" (base URL of the Cascade instance).

 required
configurationVariables dict[str, Any] | None

kwargs forwarded to the driver's cache backend (SQLiteBackend); pass an empty dict for defaults.

 required
debug dict[str, Any] | None

Optional debug config for OperationLogger (verbose nested logging); None enables normal/minimal logging.

 None
Source code in cascade_cms/wrapper.py 
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
def __init__(
    self,
    environmentVariables: EnvironmentVars,
    configurationVariables: dict[str, Any] | None,
    debug: dict[str, Any] | None = None,
):
    """Initialize the logger, driver, and operations builder.

    Args:
        environmentVariables: Must contain "SERVER" (label used in log
            output), "API_KEY" (Cascade bearer token), and
            "CASCADE_URL" (base URL of the Cascade instance).
        configurationVariables: kwargs forwarded to the driver's cache
            backend (`SQLiteBackend`); pass an empty dict for defaults.
        debug: Optional debug config for `OperationLogger` (verbose
            nested logging); None enables normal/minimal logging.
    """
    self._logger = OperationLogger(
        server=environmentVariables["SERVER"],
        debug_config=debug,
    )
    self._driver = CascadeCMSRestDriver(
        environmentVariables['API_KEY'],
        environmentVariables['CASCADE_URL'],
        configurationVariables,
        logger=self._logger,
    )
    self.operations = Operations(self._driver, _logger=self._logger)

    self._logger.log_init(
        environmentVariables['CASCADE_URL'],
        os.path.basename(sys.argv[0]),
    )

__exit__(exc_type, exc_value, traceback)

Log session exit and close the driver, then propagate exceptions.

Any exception raised inside the with block (other than RuntimeWarning) is re-raised after cleanup runs.

Source code in cascade_cms/wrapper.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def __exit__(self, exc_type, exc_value, traceback):
    """Log session exit and close the driver, then propagate exceptions.

    Any exception raised inside the `with` block (other than
    `RuntimeWarning`) is re-raised after cleanup runs.
    """
    try:
        self._logger.log_exit()
        self._driver.close()
    except Exception as e:  # noqa: BLE001 - log cleanup failure without masking the original exception
        self._logger.log_python_error(e)

    if exc_type is not None and not isinstance(exc_type, RuntimeWarning):
        return False  # Propagate the exception

submit_requests(result_type=None, *, executor=None) 

submit_requests(result_type: type[T], *, executor: Executor | None = None) -> list[T]

submit_requests(*, executor: Executor | None = None) -> list[CascadeObjects]

Run every registered operation chain and return one result per chain.

Chains run concurrently, but the nodes inside a chain run strictly in order — each operation or callback receives the previous node's result. A chain stops at its first failure without affecting any other chain.

Results line up with the chains in the order they were created, so results[0] belongs to the first chain built. Failures appear in that list rather than being dropped: a CascadeError for an API failure, or the exception object a callback raised. They are returned as values, not raised — a missed isinstance(result, CascadeError | Exception) check lets a failure flow downstream as if it were a normal result.

The chain list is cleared afterwards, so callbacks registered for one batch never run again in the next.

Parameters:

Name Type Description Default
result_type type[T] | None

Type hint for Pylance/mypy (e.g., submit_requests(Asset))

 None
executor Executor | None

Optional Executor for sync callbacks. Use ThreadPoolExecutor (default) for I/O-bound work. Use ProcessPoolExecutor(max_workers=) for CPU-bound work.

 None

Example (CPU-bound callbacks): from concurrent.futures import ProcessPoolExecutor from os import cpu_count

with ProcessPoolExecutor(max_workers=cpu_count()) as executor:
    cascade.operations.read(id).then(optimize_image)
    results = cascade.submit_requests(executor=executor)

Returns:

Type Description
list[CascadeObjects] | list[T]

One entry per chain: its final result, or the error that stopped it.
Source code in cascade_cms/wrapper.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
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
def submit_requests(self, result_type: type[T] | None = None, *, executor: Executor | None = None) -> list[CascadeObjects] | list[T]:
    """
    Run every registered operation chain and return one result per chain.

    Chains run concurrently, but the nodes inside a chain run strictly in
    order — each operation or callback receives the previous node's
    result. A chain stops at its first failure without affecting any
    other chain.

    Results line up with the chains **in the order they were created**,
    so `results[0]` belongs to the first chain built. Failures appear in
    that list rather than being dropped: a `CascadeError` for an API
    failure, or the exception object a callback raised. They are
    returned as values, not raised — a missed `isinstance(result,
    CascadeError | Exception)` check lets a failure flow downstream as
    if it were a normal result.

    The chain list is cleared afterwards, so callbacks registered for one
    batch never run again in the next.

    Args:
        result_type: Type hint for Pylance/mypy (e.g., submit_requests(Asset))
        executor: Optional Executor for sync callbacks.
                 Use ThreadPoolExecutor (default) for I/O-bound work.
                 Use ProcessPoolExecutor(max_workers=<cpu_count>) for CPU-bound work.

    Example (CPU-bound callbacks):
        from concurrent.futures import ProcessPoolExecutor
        from os import cpu_count

        with ProcessPoolExecutor(max_workers=cpu_count()) as executor:
            cascade.operations.read(id).then(optimize_image)
            results = cascade.submit_requests(executor=executor)

    Returns:
        One entry per chain: its final result, or the error that stopped it.
    """
    chains = list(self.operations._chains)
    if not chains:
        return []

    self._logger.log_batch_start()
    try:
        results = self._driver.eventLoop.run_until_complete(
            self._execute_chains(chains, executor)
        )
    except Exception as e:  # noqa: BLE001 - top-level entry point must not raise; log and return empty
        self._logger.log_python_error(e)
        return []
    finally:
        self.operations._reset_chains()

    succeeded = sum(1 for r in results if not isinstance(r, CascadeError | Exception))
    self._logger.log_batch_end(succeeded, len(chains))
    return results