Skip to content

Driver

driver

RequestExecutor dataclass

Represents a single HTTP request and how to parse its response.

Built by an OperationChain and executed concurrently, alongside the rest of its batch, via CascadeCMSRestDriver.execute_requests().

Source code in cascade_cms/driver.py 
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
@dataclass
class RequestExecutor[T]:
    """Represents a single HTTP request and how to parse its response.

    Built by an `OperationChain` and executed concurrently, alongside the
    rest of its batch, via `CascadeCMSRestDriver.execute_requests()`.
    """

    url: str
    method: Literal["GET", "POST", "PUT"]
    parser: Callable[..., ResponseParser[T]]
    payload: Payloads | None = None
    identifier: Any = None

    @property
    def log_key(self) -> str:
        """Stable key for naming this request's verbose-mode JSON files.

        Prefers the identifier's UUID hex (`IdentifierType`); falls back to
        a short hash of the URL when only a `Path` (no UUID) or no
        identifier at all is available, so a `{key}_request.json`/
        `{key}_response.json` pair still lines up for the same request.
        """
        identifier = self.identifier
        if hasattr(identifier, "get_id") and identifier.get_id:
            return str(identifier.get_id)
        return hashlib.sha1(self.url.encode()).hexdigest()[:16]

    async def fetch(
        self,
        session: ClientSession,
        sem: asyncio.Semaphore,
        logger: "OperationLogger | None" = None,
    ) -> T:
        """Execute this request and parse its response.

        Concurrency across requests is bounded by `sem`.

        Args:
            session: Shared aiohttp session to issue the request on.
            sem: Semaphore limiting concurrent in-flight requests.
            logger: Optional logger for recording request/response detail.

        Returns:
            The parsed response content (an asset, list, error, etc.).
        """
        async with sem:
            payload_bytes: bytes | None = None
            if self.payload:
                payload_bytes = serialize_payload(self.payload)

            async with session.request(
                self.method,
                self.url,
                data=payload_bytes,
            ) as response:
                if response.status != 200:
                    # Cascade always answers 200 with a JSON body; anything
                    # else is an HTML page from the Tomcat layer. Report the
                    # status only — the body is never read, logged or shown.
                    return CascadeError(  # type: ignore[return-value]
                        message=f"{response.status} {response.reason}"
                    )
                raw_data = await response.read()
                if logger:
                    logger.write_response_file(self.log_key, raw_data)
                    logger.log_network_headers(
                        dict(response.request_info.headers),
                        dict(response.headers),
                    )
                # parse raw data
                parsed_response = self.parser(raw_data)
                return parsed_response._content  # type: ignore[return-value]

log_key property

Stable key for naming this request's verbose-mode JSON files.

Prefers the identifier's UUID hex (IdentifierType); falls back to a short hash of the URL when only a Path (no UUID) or no identifier at all is available, so a {key}_request.json/ {key}_response.json pair still lines up for the same request.

fetch(session, sem, logger=None) async

Execute this request and parse its response.

Concurrency across requests is bounded by sem.

Parameters:

Name Type Description Default
session ClientSession

Shared aiohttp session to issue the request on.

 required
sem Semaphore

Semaphore limiting concurrent in-flight requests.

 required
logger OperationLogger | None

Optional logger for recording request/response detail.

 None

Returns:

Type Description
T

The parsed response content (an asset, list, error, etc.).
Source code in cascade_cms/driver.py 
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
async def fetch(
    self,
    session: ClientSession,
    sem: asyncio.Semaphore,
    logger: "OperationLogger | None" = None,
) -> T:
    """Execute this request and parse its response.

    Concurrency across requests is bounded by `sem`.

    Args:
        session: Shared aiohttp session to issue the request on.
        sem: Semaphore limiting concurrent in-flight requests.
        logger: Optional logger for recording request/response detail.

    Returns:
        The parsed response content (an asset, list, error, etc.).
    """
    async with sem:
        payload_bytes: bytes | None = None
        if self.payload:
            payload_bytes = serialize_payload(self.payload)

        async with session.request(
            self.method,
            self.url,
            data=payload_bytes,
        ) as response:
            if response.status != 200:
                # Cascade always answers 200 with a JSON body; anything
                # else is an HTML page from the Tomcat layer. Report the
                # status only — the body is never read, logged or shown.
                return CascadeError(  # type: ignore[return-value]
                    message=f"{response.status} {response.reason}"
                )
            raw_data = await response.read()
            if logger:
                logger.write_response_file(self.log_key, raw_data)
                logger.log_network_headers(
                    dict(response.request_info.headers),
                    dict(response.headers),
                )
            # parse raw data
            parsed_response = self.parser(raw_data)
            return parsed_response._content  # type: ignore[return-value]

CascadeCMSRestDriver

URL builder and executor for Cascade CMS 8 REST API. Inherits URL-building from CascadeCMSURLBuilder and applies an optional parser function to each response.

Source code in cascade_cms/driver.py 
 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
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
class CascadeCMSRestDriver:
    """
    URL builder and executor for Cascade CMS 8 REST API.
    Inherits URL-building from CascadeCMSURLBuilder and applies an optional parser function to each response.
    """

    MAX_REQUESTS = 50  # limits the requests to 50 requests per submission.

    def __init__(
        self,
        apiKey: str,
        cascade_url: str,
        logger: OperationLogger | None = None,
    ):
        """Set up the aiohttp session and event loop for this driver.

        Creates a dedicated event loop that is reused for the lifetime of
        this driver instance (rather than one per call), so all requests
        run on the same loop.

        Args:
            apiKey: Cascade CMS API bearer token.
            cascade_url: Base URL of the Cascade CMS instance (without
                the `/api/v1` suffix, which is appended automatically).
            logger: Optional logger for recording operations/errors.
        """

        self._apiKey = apiKey
        self._logger = logger

        self.base_url = f"{cascade_url}/api/v1"

        # Stores response object for debugging purpose. the ClientResponse object contains request infomation as well.
        self.request_response_info: ClientResponse | None = None

        # intializing event loop, for re-use for the rest of the session.
        self.eventLoop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.eventLoop)

        headers = {
            "Authorization": f"Bearer {self._apiKey}",
            "Cache-Control": "private",
            "Content-Type": "application/json;charset=UTF-8",
        }

        # Shared across every chain running on this driver, so concurrency
        # stays bounded no matter how many chains are in flight at once.
        self._semaphore = asyncio.Semaphore(self.MAX_REQUESTS)

        async def _create_session():
            return ClientSession(headers=headers)

        self.session = self.eventLoop.run_until_complete(_create_session())

    def _build_url(self, *segments):
        return "/".join([self.base_url, *map(str, segments)])

    async def execute_requests(
        self,
        requests: list[RequestExecutor],
    ) -> list[Any]:
        """Run the given requests concurrently and return their results in order.

        This is the single execution core: operation chains await it directly.

        Concurrency is bounded by `MAX_REQUESTS` via a driver-wide semaphore,
        so many chains running at once still cannot exceed that ceiling.
        Unlike a plain `gather`, results line up **one-for-one with `requests`**
        in submission order, and a request that raises returns the exception
        object in its slot rather than being dropped — chains rely on that to
        report which step failed.

        Args:
            requests: The requests to execute.

        Returns:
            One entry per request, in the same order: the parsed response, a
            `CascadeError`, or the `Exception` the request raised.
        """

        async def _run(executor: RequestExecutor) -> Any:
            # Deliberate boundary (see class docstring / Section 1.10 of the
            # design doc): the driver stays chain-agnostic and only writes
            # raw request/response artifacts (via RequestExecutor.fetch's
            # own logger calls) — it has no visibility into which step, in
            # which chain, a request belongs to, so rendering a chain's
            # pipeline line or its `v`/`!ERROR:` alignment is OperationChain's
            # job (execute/execute_async), done after this returns a value.
            try:
                return await executor.fetch(
                    self.session, self._semaphore, self._logger
                )
            except Exception as exc:  # noqa: BLE001 - surfaced as a value, see docstring
                return exc

        return list(await asyncio.gather(*(_run(request) for request in requests)))

    def _submitRequests(
        self,
        requests: list[RequestExecutor],
    ) -> list[Any]:
        """Run the event loop to completion for a batch of requests.

        Blocks until every request finishes.

        Args:
            requests: The requests to execute.

        Returns:
            One entry per request, in submission order.
        """
        return self.eventLoop.run_until_complete(self.execute_requests(requests))

    def close(self):
        """Tear down the aiohttp session and event loop."""
        if getattr(self, "session", None) is not None and not getattr(
            self.session, "closed", False
        ):
            self.eventLoop.run_until_complete(self.session.close())

        self.eventLoop.close()
        asyncio.set_event_loop(None)
        return True

__init__(apiKey, cascade_url, logger=None)

Set up the aiohttp session and event loop for this driver.

Creates a dedicated event loop that is reused for the lifetime of this driver instance (rather than one per call), so all requests run on the same loop.

Parameters:

Name Type Description Default
apiKey str

Cascade CMS API bearer token.

 required
cascade_url str

Base URL of the Cascade CMS instance (without the /api/v1 suffix, which is appended automatically).

 required
logger OperationLogger | None

Optional logger for recording operations/errors.

 None
Source code in cascade_cms/driver.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
def __init__(
    self,
    apiKey: str,
    cascade_url: str,
    logger: OperationLogger | None = None,
):
    """Set up the aiohttp session and event loop for this driver.

    Creates a dedicated event loop that is reused for the lifetime of
    this driver instance (rather than one per call), so all requests
    run on the same loop.

    Args:
        apiKey: Cascade CMS API bearer token.
        cascade_url: Base URL of the Cascade CMS instance (without
            the `/api/v1` suffix, which is appended automatically).
        logger: Optional logger for recording operations/errors.
    """

    self._apiKey = apiKey
    self._logger = logger

    self.base_url = f"{cascade_url}/api/v1"

    # Stores response object for debugging purpose. the ClientResponse object contains request infomation as well.
    self.request_response_info: ClientResponse | None = None

    # intializing event loop, for re-use for the rest of the session.
    self.eventLoop = asyncio.new_event_loop()
    asyncio.set_event_loop(self.eventLoop)

    headers = {
        "Authorization": f"Bearer {self._apiKey}",
        "Cache-Control": "private",
        "Content-Type": "application/json;charset=UTF-8",
    }

    # Shared across every chain running on this driver, so concurrency
    # stays bounded no matter how many chains are in flight at once.
    self._semaphore = asyncio.Semaphore(self.MAX_REQUESTS)

    async def _create_session():
        return ClientSession(headers=headers)

    self.session = self.eventLoop.run_until_complete(_create_session())

execute_requests(requests) async

Run the given requests concurrently and return their results in order.

This is the single execution core: operation chains await it directly.

Concurrency is bounded by MAX_REQUESTS via a driver-wide semaphore, so many chains running at once still cannot exceed that ceiling. Unlike a plain gather, results line up one-for-one with requests in submission order, and a request that raises returns the exception object in its slot rather than being dropped — chains rely on that to report which step failed.

Parameters:

Name Type Description Default
requests list[RequestExecutor]

The requests to execute.

 required

Returns:

Type Description
list[Any]

One entry per request, in the same order: the parsed response, a
list[Any]

CascadeError, or the Exception the request raised.
Source code in cascade_cms/driver.py 
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
async def execute_requests(
    self,
    requests: list[RequestExecutor],
) -> list[Any]:
    """Run the given requests concurrently and return their results in order.

    This is the single execution core: operation chains await it directly.

    Concurrency is bounded by `MAX_REQUESTS` via a driver-wide semaphore,
    so many chains running at once still cannot exceed that ceiling.
    Unlike a plain `gather`, results line up **one-for-one with `requests`**
    in submission order, and a request that raises returns the exception
    object in its slot rather than being dropped — chains rely on that to
    report which step failed.

    Args:
        requests: The requests to execute.

    Returns:
        One entry per request, in the same order: the parsed response, a
        `CascadeError`, or the `Exception` the request raised.
    """

    async def _run(executor: RequestExecutor) -> Any:
        # Deliberate boundary (see class docstring / Section 1.10 of the
        # design doc): the driver stays chain-agnostic and only writes
        # raw request/response artifacts (via RequestExecutor.fetch's
        # own logger calls) — it has no visibility into which step, in
        # which chain, a request belongs to, so rendering a chain's
        # pipeline line or its `v`/`!ERROR:` alignment is OperationChain's
        # job (execute/execute_async), done after this returns a value.
        try:
            return await executor.fetch(
                self.session, self._semaphore, self._logger
            )
        except Exception as exc:  # noqa: BLE001 - surfaced as a value, see docstring
            return exc

    return list(await asyncio.gather(*(_run(request) for request in requests)))

close()

Tear down the aiohttp session and event loop.

Source code in cascade_cms/driver.py 
208
209
210
211
212
213
214
215
216
217
def close(self):
    """Tear down the aiohttp session and event loop."""
    if getattr(self, "session", None) is not None and not getattr(
        self.session, "closed", False
    ):
        self.eventLoop.run_until_complete(self.session.close())

    self.eventLoop.close()
    asyncio.set_event_loop(None)
    return True