mirror of
https://github.com/hanzoai/datastore.git
synced 2026-09-26 13:55:07 +00:00
Alexey MilovidovandClaude Opus 4.8
295ced78d1
Follow-up to #105171, which added `ruff` with a deliberately narrow, bug-class-only rule set and left the unused-name and star-import rules off pending a dedicated cleanup pass. This enables that second batch in `pyproject.toml` and fixes every violation across the tree (the bulk in integration tests): - `F401` unused-import and `F841` unused-variable: removed, preserving any side-effecting right-hand side (e.g. `result = node.query(...)` becomes `node.query(...)`). Two `try: import ...` feature probes are kept with `# noqa: F401`. - `F541` f-string-missing-placeholders: dropped the redundant `f` prefix. - `F403`/`F405` star imports: every `from module import *` is expanded into explicit imports of the names actually used, taken from each name's real origin, so stdlib/third-party names import from their real source rather than through a test "convenience bundle". The now-empty `tests/integration/helpers/kafka/common_direct.py` bundle is removed and its three remaining consumers import `_VarintBytes`/`KafkaAdminClient` directly. Eliminating the star imports surfaced a real bug the wildcard had masked from `F821`: `generate_presets.py` in `test_prometheus_protocols` called an undefined `execute_instant_query_with_http_api`; fixed to the existing `execute_query_via_http_api`. No `black`/`isort` reformatting was applied (neither is enforced in the style check), so the diff is limited to the lint fixes themselves. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
138 lines
4.6 KiB
Python
138 lines
4.6 KiB
Python
import dataclasses
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from . import Artifact, Job, Workflow
|
|
from ._environment import _Environment
|
|
from .digest import Digest
|
|
from .s3 import S3
|
|
from .settings import Settings
|
|
from .utils import Utils
|
|
|
|
|
|
class Cache:
|
|
@dataclasses.dataclass
|
|
class CacheRecord:
|
|
class Type:
|
|
SUCCESS = "success"
|
|
|
|
type: str
|
|
sha: str
|
|
pr_number: int
|
|
branch: str
|
|
workflow: str = ""
|
|
|
|
def dump(self, path):
|
|
with open(path, "w", encoding="utf8") as f:
|
|
json.dump(dataclasses.asdict(self), f)
|
|
|
|
@classmethod
|
|
def from_fs(cls, path):
|
|
with open(path, "r", encoding="utf8") as f:
|
|
return Cache.CacheRecord(**json.load(f))
|
|
|
|
@classmethod
|
|
def from_dict(cls, obj):
|
|
return Cache.CacheRecord(**obj)
|
|
|
|
def __init__(self):
|
|
self.digest = Digest()
|
|
self.success = {} # type Dict[str, Any]
|
|
|
|
@classmethod
|
|
def push_success_record(
|
|
cls, job_name, job_digest, sha, workflow_name, if_not_exist
|
|
):
|
|
type_ = Cache.CacheRecord.Type.SUCCESS
|
|
record = Cache.CacheRecord(
|
|
type=type_,
|
|
sha=sha,
|
|
pr_number=_Environment.get().PR_NUMBER,
|
|
branch=_Environment.get().BRANCH,
|
|
workflow=workflow_name,
|
|
)
|
|
assert (
|
|
Settings.CACHE_S3_PATH
|
|
), "Setting CACHE_S3_PATH must be defined with enabled CI Cache"
|
|
record_path = f"{Settings.CACHE_S3_PATH}/v{Settings.CACHE_VERSION}/{Utils.normalize_string(job_name)}/{job_digest}/{type_}"
|
|
record_file = Path(Settings.TEMP_DIR) / type_
|
|
record.dump(record_file)
|
|
S3.put(
|
|
s3_path=record_path,
|
|
local_path=record_file,
|
|
if_none_matched=if_not_exist,
|
|
no_strict=True,
|
|
)
|
|
record_file.unlink()
|
|
|
|
def fetch_success(self, job_name, job_digest):
|
|
type_ = Cache.CacheRecord.Type.SUCCESS
|
|
assert (
|
|
Settings.CACHE_S3_PATH
|
|
), "Setting CACHE_S3_PATH must be defined with enabled CI Cache"
|
|
record_path = f"{Settings.CACHE_S3_PATH}/v{Settings.CACHE_VERSION}/{Utils.normalize_string(job_name)}/{job_digest}/{type_}"
|
|
record_file_local_dir = (
|
|
f"{Settings.CACHE_LOCAL_PATH}/{Utils.normalize_string(job_name)}/"
|
|
)
|
|
Path(record_file_local_dir).mkdir(parents=True, exist_ok=True)
|
|
|
|
# _skip_download_counter=True to avoid races for multithreaded downloads
|
|
res = S3.copy_file_from_s3(
|
|
s3_path=record_path,
|
|
local_path=record_file_local_dir,
|
|
_skip_download_counter=True,
|
|
no_strict=True,
|
|
)
|
|
|
|
if res:
|
|
print(f"Cache record found, job [{job_name}], digest [{job_digest}]")
|
|
self.success[job_name] = True
|
|
return Cache.CacheRecord.from_fs(Path(record_file_local_dir) / type_)
|
|
return None
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# test
|
|
c = Cache()
|
|
workflow = Workflow.Config(
|
|
name="TEST",
|
|
event=Workflow.Event.PULL_REQUEST,
|
|
jobs=[
|
|
Job.Config(
|
|
name="JobA",
|
|
runs_on=["some"],
|
|
command="python -m unittest ./ci/tests/example_1/test_example_produce_artifact.py",
|
|
provides=["greet"],
|
|
job_requirements=Job.Requirements(
|
|
python_requirements_txt="./ci/requirements.txt"
|
|
),
|
|
digest_config=Job.CacheDigestConfig(
|
|
# example: use glob to include files
|
|
include_paths=["./ci/tests/example_1/test_example_consume*.py"],
|
|
),
|
|
),
|
|
Job.Config(
|
|
name="JobB",
|
|
runs_on=["some"],
|
|
command="python -m unittest ./ci/tests/example_1/test_example_consume_artifact.py",
|
|
requires=["greet"],
|
|
job_requirements=Job.Requirements(
|
|
python_requirements_txt="./ci/requirements.txt"
|
|
),
|
|
digest_config=Job.CacheDigestConfig(
|
|
# example: use dir to include files recursively
|
|
include_paths=["./ci/tests/example_1"],
|
|
# example: use glob to exclude files from digest
|
|
exclude_paths=[
|
|
"./ci/tests/example_1/test_example_consume*",
|
|
"./**/*.pyc",
|
|
],
|
|
),
|
|
),
|
|
],
|
|
artifacts=[Artifact.Config(type="s3", name="greet", path="hello")],
|
|
enable_cache=True,
|
|
)
|
|
for job in workflow.jobs:
|
|
print(c.digest.calc_job_digest(job))
|