akamai migration#2888
Conversation
Reviewer's GuideMigrates the Akamai module from Poetry to uv and Python 3.14, updates tooling (ruff, mypy, pytest), adjusts Docker image and runtime environment accordingly, modernizes typing/timezone usage, and aligns configuration/manifest versions. File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
|
Review the following changes in direct dependencies. Learn more about Socket for GitHub.
|
There was a problem hiding this comment.
Hey - I've found 1 issue
Fixed security issues:
- black (link) · Dashboard
- cryptography (link) · Dashboard
- dulwich (link) · Dashboard
- h11 (link)
- msgpack (link) · Dashboard
- orjson (link)
- pyasn1 (link) · Dashboard
- pyopenssl (link) · Dashboard
- urllib3 (link) · Dashboard
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="Akamai/akamai_modules/connector_akamai_waf.py" line_range="140-143" />
<code_context>
event["httpMessage"]["responseHeaders"] = response_headers
- def __fetch_next_events(self, from_date: int) -> Generator[list, None, None]:
+ def __fetch_next_events(self, from_date: int) -> Generator[list]:
url = f"{self.module.configuration.base_url}/siem/v1/configs/{self.configuration.config_id}"
response = self.client.get(
</code_context>
<issue_to_address>
**suggestion:** Clarify the generator element types for better static checking and readability.
Since you’re updating the hints, please specify the yielded types more precisely. For example, `__fetch_next_events` could be `Generator[list[dict[str, Any]], None, None]`, and `fetch_events` / `filter_processed_events` `Generator[dict[str, Any], None, None]`. This helps mypy and readers understand the data shape flowing through the connector.
Suggested implementation:
```python
from cachetools import Cache, LRUCache
from pydantic import Field
from typing import Any, Generator
from sekoia_automation.checkpoint import CheckpointTimestamp, TimeUnit
from sekoia_automation.connector import Connector, DefaultConnectorConfiguration
```
```python
if "responseHeaders" in event.get("httpMessage", {}):
event["httpMessage"]["responseHeaders"] = response_headers
def __fetch_next_events(self, from_date: int) -> Generator[list[dict[str, Any]], None, None]:
```
```python
)
def fetch_events(self) -> Generator[dict[str, Any], None, None]:
most_recent_date_seen: int = self.from_timestamp
```
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| def __fetch_next_events(self, from_date: int) -> Generator[list]: | ||
| url = f"{self.module.configuration.base_url}/siem/v1/configs/{self.configuration.config_id}" | ||
| response = self.client.get( | ||
| url=url, params={"from": from_date, "limit": self.page_size}, timeout=60, stream=True |
There was a problem hiding this comment.
suggestion: Clarify the generator element types for better static checking and readability.
Since you’re updating the hints, please specify the yielded types more precisely. For example, __fetch_next_events could be Generator[list[dict[str, Any]], None, None], and fetch_events / filter_processed_events Generator[dict[str, Any], None, None]. This helps mypy and readers understand the data shape flowing through the connector.
Suggested implementation:
from cachetools import Cache, LRUCache
from pydantic import Field
from typing import Any, Generator
from sekoia_automation.checkpoint import CheckpointTimestamp, TimeUnit
from sekoia_automation.connector import Connector, DefaultConnectorConfiguration if "responseHeaders" in event.get("httpMessage", {}):
event["httpMessage"]["responseHeaders"] = response_headers
def __fetch_next_events(self, from_date: int) -> Generator[list[dict[str, Any]], None, None]: )
def fetch_events(self) -> Generator[dict[str, Any], None, None]:
most_recent_date_seen: int = self.from_timestamp
squioc
left a comment
There was a problem hiding this comment.
Thank you for the work.
It seems that some tasks are missing from the mise.toml file.
| python = "3.14" | ||
|
|
||
| [env] | ||
| _.python.venv = { path = ".venv", create = false } |
There was a problem hiding this comment.
| _.python.venv = { path = ".venv", create = false } | |
| _.python.venv = { path = ".venv", create = false } | |
| [tasks.test] | |
| description = "Run tests" | |
| run = "uv run pytest tests/" | |
| [tasks.lint] | |
| description = "Check code style and types" | |
| run = [ | |
| "uv run ruff check .", | |
| "uv run ruff format --check .", | |
| "uv run mypy .", | |
| ] | |
| [tasks.format] | |
| description = "Format and auto-fix code" | |
| run = [ | |
| "uv run ruff check --fix .", | |
| "uv run ruff format .", | |
| ] | |
| [tasks.check] | |
| description = "Check typing" | |
| run = "uvx mypy --install-types --non-interactive --ignore-missing-imports --show-column-numbers --hide-error-context ." | |
| [tasks.compliance] | |
| description = "check the compliance of the automation module" | |
| dir = "../utils" | |
| run = "uv run python compliance check --changes" | |
| [tasks.validate] | |
| description = "Check the automation module" | |
| depends = ["lint", "format", "check", "compliance"] |
Summary by Sourcery
Migrate the Akamai automation module from Poetry to uv and update it for Python 3.14, including related tooling, configuration, and minor code/test adjustments.
Enhancements:
Build:
Documentation:
Tests:
Chores: