From e0918ce34cf2cd82f92dc88d83bd11b66a8bdbb1 Mon Sep 17 00:00:00 2001 From: Daniel Burbano Date: Wed, 22 Jul 2026 17:04:28 -0500 Subject: [PATCH 1/8] feat: normalize GitHub repository metadata --- ingestion/github/repository_metadata.py | 337 ++++++++++++++++++++++++ tests/test_repository_metadata.py | 176 +++++++++++++ 2 files changed, 513 insertions(+) create mode 100644 ingestion/github/repository_metadata.py create mode 100644 tests/test_repository_metadata.py diff --git a/ingestion/github/repository_metadata.py b/ingestion/github/repository_metadata.py new file mode 100644 index 0000000..6644d04 --- /dev/null +++ b/ingestion/github/repository_metadata.py @@ -0,0 +1,337 @@ +"""Normalized repository metadata derived from GitHub API payloads.""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from datetime import UTC, datetime +from typing import Any, Self + + +class RepositoryMetadataError(ValueError): + """Raised when a GitHub repository payload is invalid.""" + + +@dataclass(frozen=True, slots=True) +class RepositoryMetadata: + """Stable representation of repository metadata for analytics.""" + + repository_id: int + repository_name: str + full_name: str + owner_login: str + description: str | None + visibility: str + default_branch: str + language: str | None + is_fork: bool + is_archived: bool + is_disabled: bool + created_at: datetime + updated_at: datetime + pushed_at: datetime | None + stars_count: int + forks_count: int + open_issues_count: int + subscribers_count: int + size_kb: int + html_url: str + extracted_at: datetime + + @classmethod + def from_github_payload( + cls, + payload: Mapping[str, Any], + *, + extracted_at: datetime | None = None, + ) -> Self: + """Create normalized metadata from a GitHub API response.""" + owner = _require_mapping( + payload.get("owner"), + field_name="owner", + ) + + visibility = _require_string( + payload, + field_name="visibility", + ) + + valid_visibilities = { + "public", + "private", + "internal", + } + + if visibility not in valid_visibilities: + raise RepositoryMetadataError( + "Field 'visibility' must be public, private or internal." + ) + + normalized_extracted_at = ( + datetime.now(UTC) + if extracted_at is None + else _normalize_aware_datetime( + extracted_at, + field_name="extracted_at", + ) + ) + + return cls( + repository_id=_require_non_negative_integer( + payload, + field_name="id", + ), + repository_name=_require_string( + payload, + field_name="name", + ), + full_name=_require_string( + payload, + field_name="full_name", + ), + owner_login=_require_string( + owner, + field_name="login", + qualified_name="owner.login", + ), + description=_optional_string( + payload, + field_name="description", + ), + visibility=visibility, + default_branch=_require_string( + payload, + field_name="default_branch", + ), + language=_optional_string( + payload, + field_name="language", + ), + is_fork=_require_boolean( + payload, + field_name="fork", + ), + is_archived=_require_boolean( + payload, + field_name="archived", + ), + is_disabled=_require_boolean( + payload, + field_name="disabled", + ), + created_at=_require_datetime( + payload, + field_name="created_at", + ), + updated_at=_require_datetime( + payload, + field_name="updated_at", + ), + pushed_at=_optional_datetime( + payload, + field_name="pushed_at", + ), + stars_count=_require_non_negative_integer( + payload, + field_name="stargazers_count", + ), + forks_count=_require_non_negative_integer( + payload, + field_name="forks_count", + ), + open_issues_count=_require_non_negative_integer( + payload, + field_name="open_issues_count", + ), + subscribers_count=_require_non_negative_integer( + payload, + field_name="subscribers_count", + ), + size_kb=_require_non_negative_integer( + payload, + field_name="size", + ), + html_url=_require_string( + payload, + field_name="html_url", + ), + extracted_at=normalized_extracted_at, + ) + + def to_record(self) -> dict[str, Any]: + """Return a stable record suitable for analytical storage.""" + return { + "repository_id": self.repository_id, + "repository_name": self.repository_name, + "full_name": self.full_name, + "owner_login": self.owner_login, + "description": self.description, + "visibility": self.visibility, + "default_branch": self.default_branch, + "language": self.language, + "is_fork": self.is_fork, + "is_archived": self.is_archived, + "is_disabled": self.is_disabled, + "created_at": self.created_at, + "updated_at": self.updated_at, + "pushed_at": self.pushed_at, + "stars_count": self.stars_count, + "forks_count": self.forks_count, + "open_issues_count": self.open_issues_count, + "subscribers_count": self.subscribers_count, + "size_kb": self.size_kb, + "html_url": self.html_url, + "extracted_at": self.extracted_at, + } + + +def _require_mapping( + value: object, + *, + field_name: str, +) -> Mapping[str, Any]: + """Validate that a payload field contains an object.""" + if not isinstance(value, Mapping): + raise RepositoryMetadataError(f"Field '{field_name}' must be a JSON object.") + + return value + + +def _require_string( + payload: Mapping[str, Any], + *, + field_name: str, + qualified_name: str | None = None, +) -> str: + """Return a required non-empty string field.""" + value = payload.get(field_name) + display_name = qualified_name or field_name + + if not isinstance(value, str) or not value.strip(): + raise RepositoryMetadataError( + f"Field '{display_name}' must be a non-empty string." + ) + + return value.strip() + + +def _optional_string( + payload: Mapping[str, Any], + *, + field_name: str, +) -> str | None: + """Return an optional string field.""" + value = payload.get(field_name) + + if value is None: + return None + + if not isinstance(value, str): + raise RepositoryMetadataError(f"Field '{field_name}' must be a string or null.") + + return value + + +def _require_boolean( + payload: Mapping[str, Any], + *, + field_name: str, +) -> bool: + """Return a required boolean field.""" + value = payload.get(field_name) + + if not isinstance(value, bool): + raise RepositoryMetadataError(f"Field '{field_name}' must be a boolean.") + + return value + + +def _require_non_negative_integer( + payload: Mapping[str, Any], + *, + field_name: str, +) -> int: + """Return a required non-negative integer field.""" + value = payload.get(field_name) + + if isinstance(value, bool) or not isinstance(value, int): + raise RepositoryMetadataError(f"Field '{field_name}' must be an integer.") + + if value < 0: + raise RepositoryMetadataError(f"Field '{field_name}' cannot be negative.") + + return value + + +def _require_datetime( + payload: Mapping[str, Any], + *, + field_name: str, +) -> datetime: + """Return a required timezone-aware datetime field.""" + value = payload.get(field_name) + + return _parse_datetime( + value, + field_name=field_name, + ) + + +def _optional_datetime( + payload: Mapping[str, Any], + *, + field_name: str, +) -> datetime | None: + """Return an optional timezone-aware datetime field.""" + value = payload.get(field_name) + + if value is None: + return None + + return _parse_datetime( + value, + field_name=field_name, + ) + + +def _parse_datetime( + value: object, + *, + field_name: str, +) -> datetime: + """Parse one ISO 8601 datetime value.""" + if not isinstance(value, str) or not value.strip(): + raise RepositoryMetadataError( + f"Field '{field_name}' must be an ISO 8601 datetime." + ) + + normalized_value = value.strip() + + if normalized_value.endswith("Z"): + normalized_value = f"{normalized_value[:-1]}+00:00" + + try: + parsed_value = datetime.fromisoformat(normalized_value) + except ValueError as exc: + raise RepositoryMetadataError( + f"Field '{field_name}' contains an invalid datetime." + ) from exc + + return _normalize_aware_datetime( + parsed_value, + field_name=field_name, + ) + + +def _normalize_aware_datetime( + value: datetime, + *, + field_name: str, +) -> datetime: + """Normalize an aware datetime to UTC.""" + if value.tzinfo is None or value.utcoffset() is None: + raise RepositoryMetadataError( + f"Field '{field_name}' must include timezone information." + ) + + return value.astimezone(UTC) diff --git a/tests/test_repository_metadata.py b/tests/test_repository_metadata.py new file mode 100644 index 0000000..97bf4a5 --- /dev/null +++ b/tests/test_repository_metadata.py @@ -0,0 +1,176 @@ +"""Tests for normalized GitHub repository metadata.""" + +from datetime import UTC, datetime + +import pytest + +from ingestion.github.repository_metadata import ( + RepositoryMetadata, + RepositoryMetadataError, +) + +VALID_PAYLOAD = { + "id": 1296269, + "name": "Hello-World", + "full_name": "octocat/Hello-World", + "owner": { + "login": "octocat", + }, + "description": "Example repository", + "visibility": "public", + "default_branch": "main", + "language": "Python", + "fork": False, + "archived": False, + "disabled": False, + "created_at": "2011-01-26T19:01:12Z", + "updated_at": "2011-01-26T19:14:43Z", + "pushed_at": "2011-01-26T19:06:43Z", + "stargazers_count": 80, + "forks_count": 9, + "open_issues_count": 2, + "subscribers_count": 42, + "size": 108, + "html_url": "https://github.com/octocat/Hello-World", +} + + +def test_from_github_payload_normalizes_repository() -> None: + """A valid GitHub payload must produce normalized metadata.""" + extracted_at = datetime( + 2026, + 7, + 22, + 21, + 45, + tzinfo=UTC, + ) + + metadata = RepositoryMetadata.from_github_payload( + VALID_PAYLOAD, + extracted_at=extracted_at, + ) + + assert metadata.repository_id == 1296269 + assert metadata.repository_name == "Hello-World" + assert metadata.full_name == "octocat/Hello-World" + assert metadata.owner_login == "octocat" + assert metadata.visibility == "public" + assert metadata.default_branch == "main" + assert metadata.language == "Python" + assert metadata.stars_count == 80 + assert metadata.forks_count == 9 + assert metadata.open_issues_count == 2 + assert metadata.subscribers_count == 42 + assert metadata.size_kb == 108 + assert metadata.extracted_at == extracted_at + + assert metadata.created_at == datetime( + 2011, + 1, + 26, + 19, + 1, + 12, + tzinfo=UTC, + ) + + +def test_to_record_returns_stable_analytical_shape() -> None: + """The normalized record must expose the expected warehouse columns.""" + metadata = RepositoryMetadata.from_github_payload( + VALID_PAYLOAD, + ) + + record = metadata.to_record() + + assert set(record) == { + "repository_id", + "repository_name", + "full_name", + "owner_login", + "description", + "visibility", + "default_branch", + "language", + "is_fork", + "is_archived", + "is_disabled", + "created_at", + "updated_at", + "pushed_at", + "stars_count", + "forks_count", + "open_issues_count", + "subscribers_count", + "size_kb", + "html_url", + "extracted_at", + } + + assert record["full_name"] == "octocat/Hello-World" + assert record["stars_count"] == 80 + + +def test_nullable_repository_fields_are_supported() -> None: + """Nullable fields returned by GitHub must remain nullable.""" + payload = { + **VALID_PAYLOAD, + "description": None, + "language": None, + "pushed_at": None, + } + + metadata = RepositoryMetadata.from_github_payload(payload) + + assert metadata.description is None + assert metadata.language is None + assert metadata.pushed_at is None + + +def test_missing_owner_login_is_rejected() -> None: + """A repository without an owner login must be rejected.""" + payload = { + **VALID_PAYLOAD, + "owner": {}, + } + + with pytest.raises( + RepositoryMetadataError, + match="owner.login", + ): + RepositoryMetadata.from_github_payload(payload) + + +def test_negative_repository_metric_is_rejected() -> None: + """Repository counters cannot contain negative values.""" + payload = { + **VALID_PAYLOAD, + "stargazers_count": -1, + } + + with pytest.raises( + RepositoryMetadataError, + match="stargazers_count.*cannot be negative", + ): + RepositoryMetadata.from_github_payload(payload) + + +def test_naive_extraction_datetime_is_rejected() -> None: + """The extraction timestamp must include timezone information.""" + naive_datetime = datetime( + 2026, + 7, + 22, + 21, + 45, + ) + + with pytest.raises( + RepositoryMetadataError, + match="extracted_at.*timezone", + ): + RepositoryMetadata.from_github_payload( + VALID_PAYLOAD, + extracted_at=naive_datetime, + ) From 14bed83942517c81f802703b6ca74ac7c3d5306f Mon Sep 17 00:00:00 2001 From: Daniel Burbano Date: Wed, 22 Jul 2026 17:08:42 -0500 Subject: [PATCH 2/8] feat: add GitHub repository extraction service --- ingestion/github/repository_service.py | 107 +++++++++++++++++++++ scripts/check_repository_extraction.py | 62 ++++++++++++ tests/test_repository_service.py | 128 +++++++++++++++++++++++++ 3 files changed, 297 insertions(+) create mode 100644 ingestion/github/repository_service.py create mode 100644 scripts/check_repository_extraction.py create mode 100644 tests/test_repository_service.py diff --git a/ingestion/github/repository_service.py b/ingestion/github/repository_service.py new file mode 100644 index 0000000..238bfa8 --- /dev/null +++ b/ingestion/github/repository_service.py @@ -0,0 +1,107 @@ +"""Service for extracting normalized GitHub repository metadata.""" + +from __future__ import annotations + +import logging +from datetime import datetime + +from ingestion.github.client import GitHubClient +from ingestion.github.repository_metadata import ( + RepositoryMetadata, + RepositoryMetadataError, +) + +logger = logging.getLogger(__name__) + + +class RepositoryExtractionError(RuntimeError): + """Raised when repository metadata cannot be extracted or normalized.""" + + +class GitHubRepositoryService: + """Extract and normalize repository metadata from GitHub.""" + + def __init__(self, client: GitHubClient) -> None: + """Initialize the service with a reusable GitHub API client.""" + self._client = client + + def extract_repository( + self, + owner: str, + repository: str, + *, + extracted_at: datetime | None = None, + ) -> RepositoryMetadata: + """Extract and normalize metadata for one GitHub repository.""" + normalized_owner = _normalize_path_component( + owner, + field_name="owner", + ) + + normalized_repository = _normalize_path_component( + repository, + field_name="repository", + ) + + endpoint = f"/repos/{normalized_owner}/{normalized_repository}" + + logger.info( + "Starting GitHub repository metadata extraction.", + extra={ + "operation": "github_repository_extraction", + "owner": normalized_owner, + "repository": normalized_repository, + "endpoint": endpoint, + }, + ) + + payload = self._client.get(endpoint) + + if not isinstance(payload, dict): + raise RepositoryExtractionError( + "GitHub repository endpoint returned a non-object JSON response." + ) + + try: + metadata = RepositoryMetadata.from_github_payload( + payload, + extracted_at=extracted_at, + ) + except RepositoryMetadataError as exc: + raise RepositoryExtractionError( + "GitHub repository metadata could not be normalized " + f"for {normalized_owner}/{normalized_repository}." + ) from exc + + logger.info( + "GitHub repository metadata extraction completed.", + extra={ + "operation": "github_repository_extraction", + "repository_id": metadata.repository_id, + "full_name": metadata.full_name, + "owner": metadata.owner_login, + "visibility": metadata.visibility, + }, + ) + + return metadata + + +def _normalize_path_component( + value: str, + *, + field_name: str, +) -> str: + """Validate one GitHub repository path component.""" + if not isinstance(value, str): + raise ValueError(f"{field_name} must be a string.") + + normalized_value = value.strip() + + if not normalized_value: + raise ValueError(f"{field_name} cannot be empty.") + + if "/" in normalized_value or "\\" in normalized_value: + raise ValueError(f"{field_name} cannot contain path separators.") + + return normalized_value diff --git a/scripts/check_repository_extraction.py b/scripts/check_repository_extraction.py new file mode 100644 index 0000000..fc659c1 --- /dev/null +++ b/scripts/check_repository_extraction.py @@ -0,0 +1,62 @@ +"""Validate repository metadata extraction against the GitHub API.""" + +from ingestion.common.config import ConfigurationError, Settings +from ingestion.common.logging import configure_logging +from ingestion.github.client import GitHubClient +from ingestion.github.exceptions import GitHubAPIError +from ingestion.github.repository_service import ( + GitHubRepositoryService, + RepositoryExtractionError, +) + + +def main() -> int: + """Extract and display normalized repository metadata.""" + try: + settings = Settings.from_env() + configure_logging(settings.log_level) + + client = GitHubClient( + token=settings.github_token, + base_url=settings.github_api_base_url, + api_version=settings.github_api_version, + timeout_seconds=settings.github_timeout_seconds, + max_retries=settings.github_max_retries, + backoff_seconds=settings.github_backoff_seconds, + ) + + service = GitHubRepositoryService(client) + + metadata = service.extract_repository( + owner="apache", + repository="airflow", + ) + + except ( + ConfigurationError, + GitHubAPIError, + RepositoryExtractionError, + ValueError, + ) as exc: + print(f"[ERROR] {exc}") + return 1 + + print("DevFlow Intelligence - Repository Extraction Check") + print("--------------------------------------------------") + print("[SUCCESS] Repository metadata extracted.") + print(f"Repository ID: {metadata.repository_id}") + print(f"Repository: {metadata.full_name}") + print(f"Owner: {metadata.owner_login}") + print(f"Visibility: {metadata.visibility}") + print(f"Default branch: {metadata.default_branch}") + print(f"Language: {metadata.language}") + print(f"Stars: {metadata.stars_count}") + print(f"Forks: {metadata.forks_count}") + print(f"Open issues: {metadata.open_issues_count}") + print(f"Extracted at: {metadata.extracted_at.isoformat()}") + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_repository_service.py b/tests/test_repository_service.py new file mode 100644 index 0000000..0d776b0 --- /dev/null +++ b/tests/test_repository_service.py @@ -0,0 +1,128 @@ +"""Tests for the GitHub repository extraction service.""" + +from datetime import UTC, datetime +from unittest.mock import Mock + +import pytest + +from ingestion.github.client import GitHubClient +from ingestion.github.repository_service import ( + GitHubRepositoryService, + RepositoryExtractionError, +) + + +def build_repository_payload() -> dict[str, object]: + """Build a valid simulated GitHub repository response.""" + return { + "id": 1296269, + "name": "airflow", + "full_name": "apache/airflow", + "owner": { + "login": "apache", + }, + "description": "Platform to programmatically author workflows", + "visibility": "public", + "default_branch": "main", + "language": "Python", + "fork": False, + "archived": False, + "disabled": False, + "created_at": "2015-04-13T18:04:58Z", + "updated_at": "2026-07-22T20:00:00Z", + "pushed_at": "2026-07-22T19:30:00Z", + "stargazers_count": 40000, + "forks_count": 15000, + "open_issues_count": 1000, + "subscribers_count": 700, + "size": 250000, + "html_url": "https://github.com/apache/airflow", + } + + +def test_extract_repository_returns_normalized_metadata() -> None: + """A valid GitHub response must be normalized by the service.""" + client = Mock(spec=GitHubClient) + client.get.return_value = build_repository_payload() + + service = GitHubRepositoryService(client) + + extracted_at = datetime( + 2026, + 7, + 22, + 22, + 0, + tzinfo=UTC, + ) + + metadata = service.extract_repository( + owner=" apache ", + repository=" airflow ", + extracted_at=extracted_at, + ) + + client.get.assert_called_once_with("/repos/apache/airflow") + + assert metadata.repository_id == 1296269 + assert metadata.full_name == "apache/airflow" + assert metadata.owner_login == "apache" + assert metadata.default_branch == "main" + assert metadata.language == "Python" + assert metadata.extracted_at == extracted_at + + +def test_extract_repository_rejects_non_object_response() -> None: + """The repository endpoint must return a JSON object.""" + client = Mock(spec=GitHubClient) + client.get.return_value = [] + + service = GitHubRepositoryService(client) + + with pytest.raises( + RepositoryExtractionError, + match="non-object JSON response", + ): + service.extract_repository( + owner="apache", + repository="airflow", + ) + + +def test_extract_repository_wraps_normalization_error() -> None: + """Invalid GitHub metadata must produce an extraction error.""" + invalid_payload = build_repository_payload() + invalid_payload["owner"] = {} + + client = Mock(spec=GitHubClient) + client.get.return_value = invalid_payload + + service = GitHubRepositoryService(client) + + with pytest.raises( + RepositoryExtractionError, + match="could not be normalized", + ) as exception_info: + service.extract_repository( + owner="apache", + repository="airflow", + ) + + assert exception_info.value.__cause__ is not None + + +def test_extract_repository_rejects_path_separator() -> None: + """Repository path components cannot contain separators.""" + client = Mock(spec=GitHubClient) + service = GitHubRepositoryService(client) + + with pytest.raises( + ValueError, + match="owner cannot contain path separators", + ): + service.extract_repository( + owner="apache/team", + repository="airflow", + ) + + client.get.assert_not_called() From eb3b914347983bf9fd06ca47239e55e728749e17 Mon Sep 17 00:00:00 2001 From: Daniel Burbano Date: Thu, 23 Jul 2026 15:43:11 -0500 Subject: [PATCH 3/8] chore: add Codex project instructions --- AGENTS.md | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) create mode 100644 AGENTS.md diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..f30754d --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,24 @@ +# DevFlow Intelligence + +- Este es un proyecto Python de Data Engineering llamado DevFlow Intelligence. +- El sistema operativo es Windows; usa exclusivamente comandos PowerShell. +- Trabaja solamente en el ticket DFI solicitado. +- No leas, modifiques, muestres ni versiones `.env`. +- No expongas tokens, credenciales ni encabezados `Authorization`. +- No modifiques archivos ajenos al alcance solicitado. +- Nunca usar el Python global del sistema. +- Nunca instalar paquetes globalmente. +- No ejecutar `pip install` sin autorización explícita. +- No instales dependencias sin autorización. +- No hagas commit, push, merge ni elimines ramas sin autorización explícita. +- Durante el desarrollo, ejecuta primero pruebas específicas y una sola vez la suite completa al final. +- Antes de ejecutar pruebas, comprobar: + - `.\.venv\Scripts\python.exe --version` +- No escanear `.venv`, `.git`, `__pycache__` ni archivos generados. +- Antes de terminar, ejecuta: + - `.\.venv\Scripts\python.exe -m py_compile RUTA_DEL_ARCHIVO` + - `.\.venv\Scripts\python.exe -m ruff format .` + - `.\.venv\Scripts\python.exe -m ruff check .` + - `.\.venv\Scripts\python.exe -m ruff format --check .` + - `.\.venv\Scripts\python.exe -m pytest` +- Da respuestas finales breves que incluyan archivos cambiados, implementación, pruebas y riesgos. From 729cf0e42bed2b816802300dd381771193f9220f Mon Sep 17 00:00:00 2001 From: Daniel Burbano Date: Thu, 23 Jul 2026 19:43:53 -0500 Subject: [PATCH 4/8] feat: add batch repository extraction and local persistence --- .gitignore | 4 +- config/repositories.example.json | 12 + ingestion/github/repository_batch.py | 480 +++++++++++++++++++++++ ingestion/github/repository_service.py | 31 +- scripts/extract_repositories.py | 85 +++++ tests/test_extract_repositories.py | 258 +++++++++++++ tests/test_repository_batch.py | 506 +++++++++++++++++++++++++ tests/test_repository_service.py | 18 + 8 files changed, 1392 insertions(+), 2 deletions(-) create mode 100644 config/repositories.example.json create mode 100644 ingestion/github/repository_batch.py create mode 100644 scripts/extract_repositories.py create mode 100644 tests/test_extract_repositories.py create mode 100644 tests/test_repository_batch.py diff --git a/.gitignore b/.gitignore index 7d91a3f..6048d3a 100644 --- a/.gitignore +++ b/.gitignore @@ -20,6 +20,8 @@ service-account.json # Datos locales data/raw/* !data/raw/.gitkeep +data/normalized/ +data/manifests/ # Airflow airflow.db @@ -43,4 +45,4 @@ Thumbs.db # Archivos temporales *.log -*.tmp \ No newline at end of file +*.tmp diff --git a/config/repositories.example.json b/config/repositories.example.json new file mode 100644 index 0000000..0f22103 --- /dev/null +++ b/config/repositories.example.json @@ -0,0 +1,12 @@ +{ + "repositories": [ + { + "owner": "apache", + "name": "airflow" + }, + { + "owner": "pandas-dev", + "name": "pandas" + } + ] +} diff --git a/ingestion/github/repository_batch.py b/ingestion/github/repository_batch.py new file mode 100644 index 0000000..0323715 --- /dev/null +++ b/ingestion/github/repository_batch.py @@ -0,0 +1,480 @@ +"""Configurable repository batch extraction with atomic local persistence. + +Each output file is published atomically from a temporary file in its own +directory. The three publications cannot form one cross-file transaction, so a +filesystem failure between replacements may leave an earlier output published. +The manifest is always published last and only after raw and normalized data. +""" + +from __future__ import annotations + +import json +import logging +import os +import tempfile +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass +from datetime import UTC, datetime +from json import JSONDecodeError +from pathlib import Path +from typing import Any, Literal + +from ingestion.github.exceptions import ( + GitHubAPIError, + GitHubAuthenticationError, + GitHubNotFoundError, + GitHubRateLimitError, +) +from ingestion.github.repository_service import ( + GitHubRepositoryService, + RepositoryExtractionError, +) + +logger = logging.getLogger(__name__) + +type BatchStatus = Literal["success", "partial_success", "failed"] +type Clock = Callable[[], datetime] + + +class RepositoryBatchConfigurationError(ValueError): + """Raised when a repository batch configuration is invalid.""" + + +@dataclass(frozen=True, slots=True) +class RepositorySpec: + """One validated GitHub repository identifier.""" + + owner: str + name: str + + +@dataclass(frozen=True, slots=True) +class RepositoryBatchConfig: + """Validated repository batch configuration.""" + + repositories: tuple[RepositorySpec, ...] + + +@dataclass(frozen=True, slots=True) +class RepositoryBatchResult: + """Summary and output locations for a completed batch run.""" + + run_id: str + status: BatchStatus + repositories_succeeded: int + repositories_failed: int + output_root: Path + manifest_path: Path + + +@dataclass(frozen=True, slots=True) +class _OutputPaths: + raw: Path + normalized: Path + manifest: Path + + +def load_repository_config(path: Path) -> RepositoryBatchConfig: + """Load and validate a repository list from a JSON file.""" + try: + with path.open(encoding="utf-8") as config_file: + document = json.load(config_file) + except JSONDecodeError as exc: + raise RepositoryBatchConfigurationError( + f"Repository configuration contains invalid JSON at line {exc.lineno}." + ) from exc + except UnicodeError as exc: + raise RepositoryBatchConfigurationError( + "Repository configuration must be valid UTF-8." + ) from exc + except OSError as exc: + raise RepositoryBatchConfigurationError( + "Repository configuration could not be read." + ) from exc + + if not isinstance(document, dict): + raise RepositoryBatchConfigurationError( + "Repository configuration root must be a JSON object." + ) + + repositories = document.get("repositories") + + if not isinstance(repositories, list) or not repositories: + raise RepositoryBatchConfigurationError( + "'repositories' must be a non-empty list." + ) + + validated: list[RepositorySpec] = [] + seen: set[tuple[str, str]] = set() + + for index, item in enumerate(repositories, start=1): + if not isinstance(item, dict): + raise RepositoryBatchConfigurationError( + f"Repository item {index} must be a JSON object." + ) + + owner = _validate_repository_component( + item.get("owner"), + field_name="owner", + item_number=index, + ) + name = _validate_repository_component( + item.get("name"), + field_name="name", + item_number=index, + ) + identity = (owner.casefold(), name.casefold()) + + if identity in seen: + raise RepositoryBatchConfigurationError( + f"Repository item {index} is a duplicate: {owner}/{name}." + ) + + seen.add(identity) + validated.append(RepositorySpec(owner=owner, name=name)) + + return RepositoryBatchConfig(repositories=tuple(validated)) + + +class GitHubRepositoryBatch: + """Extract configured repositories and persist one local batch run.""" + + def __init__( + self, + service: GitHubRepositoryService, + *, + clock: Clock | None = None, + ) -> None: + """Initialize the batch with an extraction service and injectable clock.""" + self._service = service + self._clock = clock or _utc_now + + def run( + self, + config: RepositoryBatchConfig, + *, + output_root: Path, + ) -> RepositoryBatchResult: + """Execute one repository batch and atomically publish its outputs.""" + started_at = _read_utc_clock(self._clock, field_name="started_at") + extracted_at = _read_utc_clock(self._clock, field_name="extracted_at") + run_id = extracted_at.strftime("%Y%m%dT%H%M%S%fZ") + output_root = Path(output_root) + paths = _build_output_paths( + output_root=output_root, + extraction_date=extracted_at.date().isoformat(), + run_id=run_id, + ) + _reject_existing_outputs(paths) + + raw_records: list[dict[str, Any]] = [] + normalized_records: list[dict[str, Any]] = [] + errors: list[dict[str, str]] = [] + + for repository in config.repositories: + try: + result = self._service.extract_repository_with_payload( + repository.owner, + repository.name, + extracted_at=extracted_at, + ) + raw_record = { + "run_id": run_id, + "extracted_at": extracted_at.isoformat(), + "owner": repository.owner, + "repository": repository.name, + "endpoint": result.endpoint, + "payload": result.payload, + } + normalized_record = _serialize_normalized_record( + result.metadata.to_record(), + run_id=run_id, + ) + except Exception as exc: + error = _safe_repository_error(repository, exc) + errors.append(error) + logger.error( + "GitHub repository batch item failed.", + extra={ + "operation": "github_repository_batch_item", + "owner": repository.owner, + "repository": repository.name, + "error_type": error["error_type"], + }, + ) + continue + + raw_records.append(raw_record) + normalized_records.append(normalized_record) + + succeeded = len(raw_records) + failed = len(errors) + status = _batch_status(succeeded=succeeded, failed=failed) + manifest_fields = { + "run_id": run_id, + "started_at": started_at.isoformat(), + "extracted_at": extracted_at.isoformat(), + "status": status, + "repositories_requested": len(config.repositories), + "repositories_succeeded": succeeded, + "repositories_failed": failed, + "raw_output": paths.raw.relative_to(output_root).as_posix(), + "normalized_output": paths.normalized.relative_to(output_root).as_posix(), + "errors": errors, + } + + _publish_outputs_atomically( + paths=paths, + raw_records=raw_records, + normalized_records=normalized_records, + manifest_fields=manifest_fields, + run_id=run_id, + clock=self._clock, + ) + + return RepositoryBatchResult( + run_id=run_id, + status=status, + repositories_succeeded=succeeded, + repositories_failed=failed, + output_root=output_root, + manifest_path=paths.manifest, + ) + + +def _validate_repository_component( + value: object, + *, + field_name: str, + item_number: int, +) -> str: + """Validate one owner or repository name from configuration.""" + if not isinstance(value, str) or not value.strip(): + raise RepositoryBatchConfigurationError( + f"Repository item {item_number} field '{field_name}' " + "must be a non-empty string." + ) + + normalized = value.strip() + + if "/" in normalized or "\\" in normalized: + raise RepositoryBatchConfigurationError( + f"Repository item {item_number} field '{field_name}' " + "cannot contain a path separator." + ) + + return normalized + + +def _utc_now() -> datetime: + """Return the current timezone-aware UTC time.""" + return datetime.now(UTC) + + +def _read_utc_clock(clock: Clock, *, field_name: str) -> datetime: + """Read and normalize one timezone-aware value from the batch clock.""" + value = clock() + + if not isinstance(value, datetime): + raise ValueError(f"{field_name} clock value must be a datetime.") + + if value.tzinfo is None or value.utcoffset() is None: + raise ValueError(f"{field_name} clock value must include timezone information.") + + return value.astimezone(UTC) + + +def _build_output_paths( + *, + output_root: Path, + extraction_date: str, + run_id: str, +) -> _OutputPaths: + """Build final locations for all outputs in one run.""" + partition = Path(f"extraction_date={extraction_date}") / f"run_id={run_id}" + return _OutputPaths( + raw=output_root + / "raw" + / "github" + / "repositories" + / partition + / "repositories.jsonl", + normalized=output_root + / "normalized" + / "github" + / "repositories" + / partition + / "repositories.jsonl", + manifest=output_root + / "manifests" + / "github" + / "repositories" + / partition + / "manifest.json", + ) + + +def _reject_existing_outputs(paths: _OutputPaths) -> None: + """Reject a run identifier whose final output already exists.""" + existing = [ + path for path in (paths.raw, paths.normalized, paths.manifest) if path.exists() + ] + + if existing: + raise FileExistsError(f"Batch output already exists: {existing[0]}") + + +def _safe_repository_error( + repository: RepositorySpec, + error: Exception, +) -> dict[str, str]: + """Build a manifest error without copying potentially sensitive content.""" + if isinstance(error, GitHubAuthenticationError): + error_type = type(error).__name__ + message = "GitHub authentication failed." + elif isinstance(error, GitHubNotFoundError): + error_type = type(error).__name__ + message = "GitHub repository was not found." + elif isinstance(error, GitHubRateLimitError): + error_type = type(error).__name__ + message = "GitHub API rate limit reached." + elif isinstance(error, RepositoryExtractionError): + error_type = type(error).__name__ + message = "GitHub repository response could not be normalized." + elif isinstance(error, GitHubAPIError): + error_type = type(error).__name__ + message = "GitHub API request failed." + else: + error_type = "UnexpectedRepositoryError" + message = "Unexpected repository extraction error." + + return { + "owner": repository.owner, + "repository": repository.name, + "error_type": error_type, + "message": message, + } + + +def _serialize_normalized_record( + record: Mapping[str, Any], + *, + run_id: str, +) -> dict[str, Any]: + """Copy a normalized record and convert its datetimes to UTC strings.""" + serialized: dict[str, Any] = {"run_id": run_id} + + for key, value in record.items(): + if isinstance(value, datetime): + if value.tzinfo is None or value.utcoffset() is None: + raise ValueError( + f"Normalized datetime field '{key}' must include " + "timezone information." + ) + serialized[key] = value.astimezone(UTC).isoformat() + else: + serialized[key] = value + + return serialized + + +def _batch_status(*, succeeded: int, failed: int) -> BatchStatus: + """Determine the final status from repository counters.""" + if succeeded and failed: + return "partial_success" + if succeeded: + return "success" + return "failed" + + +def _publish_outputs_atomically( + *, + paths: _OutputPaths, + raw_records: Sequence[Mapping[str, Any]], + normalized_records: Sequence[Mapping[str, Any]], + manifest_fields: Mapping[str, Any], + run_id: str, + clock: Clock, +) -> None: + """Publish raw and normalized, then timestamp and publish the manifest. + + Raw and normalized are written to local temporaries and published first. + The completion time is then read before the manifest is written and + published last. + """ + temporary_paths: list[Path] = [] + + try: + raw_temporary = _create_temporary_path( + destination=paths.raw, + run_id=run_id, + ) + temporary_paths.append(raw_temporary) + normalized_temporary = _create_temporary_path( + destination=paths.normalized, + run_id=run_id, + ) + temporary_paths.append(normalized_temporary) + manifest_temporary = _create_temporary_path( + destination=paths.manifest, + run_id=run_id, + ) + temporary_paths.append(manifest_temporary) + + _reject_existing_outputs(paths) + _write_json_lines_temporary(raw_temporary, raw_records) + _write_json_lines_temporary(normalized_temporary, normalized_records) + os.replace(raw_temporary, paths.raw) + os.replace(normalized_temporary, paths.normalized) + + completed_at = _read_utc_clock(clock, field_name="completed_at") + manifest = { + **manifest_fields, + "completed_at": completed_at.isoformat(), + } + _write_json_temporary(manifest_temporary, manifest) + os.replace(manifest_temporary, paths.manifest) + finally: + for temporary_path in temporary_paths: + temporary_path.unlink(missing_ok=True) + + +def _create_temporary_path( + *, + destination: Path, + run_id: str, +) -> Path: + """Create and close one tracked temporary beside its destination.""" + destination.parent.mkdir(parents=True, exist_ok=True) + file_descriptor, temporary_name = tempfile.mkstemp( + prefix=f".{destination.name}.{run_id}.", + suffix=".tmp", + dir=destination.parent, + ) + os.close(file_descriptor) + return Path(temporary_name) + + +def _write_json_lines_temporary( + temporary_path: Path, + records: Sequence[Mapping[str, Any]], +) -> None: + """Write and close one complete JSON Lines temporary.""" + with temporary_path.open( + mode="w", encoding="utf-8", newline="\n" + ) as temporary_file: + for record in records: + temporary_file.write(json.dumps(record, ensure_ascii=False)) + temporary_file.write("\n") + + +def _write_json_temporary( + temporary_path: Path, + document: Mapping[str, Any], +) -> None: + """Write and close one complete JSON document temporary.""" + with temporary_path.open( + mode="w", encoding="utf-8", newline="\n" + ) as temporary_file: + json.dump(document, temporary_file, ensure_ascii=False, indent=2) + temporary_file.write("\n") diff --git a/ingestion/github/repository_service.py b/ingestion/github/repository_service.py index 238bfa8..884a8e4 100644 --- a/ingestion/github/repository_service.py +++ b/ingestion/github/repository_service.py @@ -3,7 +3,9 @@ from __future__ import annotations import logging +from dataclasses import dataclass from datetime import datetime +from typing import Any from ingestion.github.client import GitHubClient from ingestion.github.repository_metadata import ( @@ -18,6 +20,15 @@ class RepositoryExtractionError(RuntimeError): """Raised when repository metadata cannot be extracted or normalized.""" +@dataclass(frozen=True, slots=True) +class RepositoryExtractionResult: + """Raw and normalized results produced by one GitHub request.""" + + endpoint: str + payload: dict[str, Any] + metadata: RepositoryMetadata + + class GitHubRepositoryService: """Extract and normalize repository metadata from GitHub.""" @@ -33,6 +44,20 @@ def extract_repository( extracted_at: datetime | None = None, ) -> RepositoryMetadata: """Extract and normalize metadata for one GitHub repository.""" + return self.extract_repository_with_payload( + owner, + repository, + extracted_at=extracted_at, + ).metadata + + def extract_repository_with_payload( + self, + owner: str, + repository: str, + *, + extracted_at: datetime | None = None, + ) -> RepositoryExtractionResult: + """Extract raw and normalized metadata with a single GitHub request.""" normalized_owner = _normalize_path_component( owner, field_name="owner", @@ -84,7 +109,11 @@ def extract_repository( }, ) - return metadata + return RepositoryExtractionResult( + endpoint=endpoint, + payload=payload, + metadata=metadata, + ) def _normalize_path_component( diff --git a/scripts/extract_repositories.py b/scripts/extract_repositories.py new file mode 100644 index 0000000..69d5bb3 --- /dev/null +++ b/scripts/extract_repositories.py @@ -0,0 +1,85 @@ +"""Run configurable GitHub repository extraction and local persistence.""" + +from __future__ import annotations + +import argparse +from collections.abc import Sequence +from pathlib import Path + +from ingestion.common.config import ConfigurationError, Settings +from ingestion.common.logging import configure_logging +from ingestion.github.client import GitHubClient +from ingestion.github.repository_batch import ( + GitHubRepositoryBatch, + RepositoryBatchConfigurationError, + load_repository_config, +) +from ingestion.github.repository_service import GitHubRepositoryService + + +def build_parser() -> argparse.ArgumentParser: + """Build the repository batch command-line parser.""" + parser = argparse.ArgumentParser( + description="Extract configured GitHub repositories to local JSON files." + ) + parser.add_argument( + "--config", + required=True, + type=Path, + help="Path to a repository configuration JSON file.", + ) + parser.add_argument( + "--output-root", + required=True, + type=Path, + help="Root directory for raw, normalized and manifest outputs.", + ) + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + """Execute the repository batch and return its process exit code.""" + args = build_parser().parse_args(argv) + + try: + config = load_repository_config(args.config) + settings = Settings.from_env() + configure_logging(settings.log_level) + + client = GitHubClient( + token=settings.github_token, + base_url=settings.github_api_base_url, + api_version=settings.github_api_version, + timeout_seconds=settings.github_timeout_seconds, + max_retries=settings.github_max_retries, + backoff_seconds=settings.github_backoff_seconds, + ) + service = GitHubRepositoryService(client) + result = GitHubRepositoryBatch(service).run( + config, + output_root=args.output_root, + ) + except (ConfigurationError, RepositoryBatchConfigurationError) as exc: + print(f"[ERROR] {exc}") + return 1 + except OSError: + print("[ERROR] Local repository batch output could not be written.") + return 1 + + print("DevFlow Intelligence - Repository Batch Extraction") + print("--------------------------------------------------") + print(f"Run ID: {result.run_id}") + print(f"Status: {result.status}") + print(f"Repositories succeeded: {result.repositories_succeeded}") + print(f"Repositories failed: {result.repositories_failed}") + print(f"Manifest: {result.manifest_path}") + + return { + "success": 0, + "partial_success": 2, + "failed": 1, + }[result.status] + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_extract_repositories.py b/tests/test_extract_repositories.py new file mode 100644 index 0000000..e4a00d6 --- /dev/null +++ b/tests/test_extract_repositories.py @@ -0,0 +1,258 @@ +"""Tests for the repository batch command-line interface.""" + +from __future__ import annotations + +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from ingestion.github.repository_batch import RepositoryBatchConfigurationError +from scripts import extract_repositories + + +@pytest.mark.parametrize( + ("status", "expected_code"), + [ + ("success", 0), + ("partial_success", 2), + ("failed", 1), + ], +) +def test_main_returns_status_code_without_real_github_calls( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + status: str, + expected_code: int, +) -> None: + """The CLI must compose dependencies and map every batch status.""" + config = object() + settings = SimpleNamespace( + github_token="test-token", + github_api_base_url="https://example.invalid", + github_api_version="test-version", + github_timeout_seconds=1.0, + github_max_retries=0, + github_backoff_seconds=0.0, + log_level="INFO", + ) + created: dict[str, object] = {} + + monkeypatch.setattr( + extract_repositories.Settings, + "from_env", + classmethod(lambda cls: settings), + ) + monkeypatch.setattr( + extract_repositories, + "configure_logging", + lambda level: created.update(log_level=level), + ) + monkeypatch.setattr( + extract_repositories, + "load_repository_config", + lambda path: created.update(config_path=path) or config, + ) + + class FakeClient: + def __init__(self, **kwargs: object) -> None: + created["client_kwargs"] = kwargs + + class FakeService: + def __init__(self, client: object) -> None: + created["client"] = client + + class FakeBatch: + def __init__(self, service: object) -> None: + created["service"] = service + + def run(self, batch_config: object, *, output_root: Path): + created["batch_config"] = batch_config + created["output_root"] = output_root + return SimpleNamespace( + run_id="20260723T120000000000Z", + status=status, + repositories_succeeded=1 if status != "failed" else 0, + repositories_failed=0 if status == "success" else 1, + manifest_path=output_root / "manifest.json", + ) + + monkeypatch.setattr(extract_repositories, "GitHubClient", FakeClient) + monkeypatch.setattr(extract_repositories, "GitHubRepositoryService", FakeService) + monkeypatch.setattr(extract_repositories, "GitHubRepositoryBatch", FakeBatch) + + config_path = tmp_path / "repositories.json" + output_root = tmp_path / "data" + exit_code = extract_repositories.main( + ["--config", str(config_path), "--output-root", str(output_root)] + ) + + assert exit_code == expected_code + assert created["config_path"] == config_path + assert created["output_root"] == output_root + assert created["batch_config"] is config + assert created["log_level"] == "INFO" + assert created["client_kwargs"] == { + "token": "test-token", + "base_url": "https://example.invalid", + "api_version": "test-version", + "timeout_seconds": 1.0, + "max_retries": 0, + "backoff_seconds": 0.0, + } + + +def test_main_returns_one_for_invalid_configuration_without_creating_client( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """Invalid local configuration must be checked before settings or clients.""" + client_created = False + settings_loaded = False + + def unexpected_settings(cls: type) -> None: + nonlocal settings_loaded + settings_loaded = True + raise AssertionError("Settings.from_env must not be called") + + monkeypatch.setattr( + extract_repositories.Settings, + "from_env", + classmethod(unexpected_settings), + ) + monkeypatch.setattr(extract_repositories, "configure_logging", lambda level: None) + + def invalid_config(path: Path): + raise RepositoryBatchConfigurationError("Invalid repository configuration.") + + class UnexpectedClient: + def __init__(self, **kwargs: object) -> None: + nonlocal client_created + client_created = True + + monkeypatch.setattr(extract_repositories, "load_repository_config", invalid_config) + monkeypatch.setattr(extract_repositories, "GitHubClient", UnexpectedClient) + + exit_code = extract_repositories.main( + [ + "--config", + str(tmp_path / "invalid.json"), + "--output-root", + str(tmp_path / "data"), + ] + ) + + assert exit_code == 1 + assert client_created is False + assert settings_loaded is False + + +def test_main_reports_missing_configuration_as_input_error( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """A missing input file must not be described as an output write failure.""" + settings_loaded = False + + def unexpected_settings(cls: type) -> None: + nonlocal settings_loaded + settings_loaded = True + raise AssertionError("Settings.from_env must not be called") + + monkeypatch.setattr( + extract_repositories.Settings, + "from_env", + classmethod(unexpected_settings), + ) + + exit_code = extract_repositories.main( + [ + "--config", + str(tmp_path / "missing.json"), + "--output-root", + str(tmp_path / "data"), + ] + ) + + output = capsys.readouterr().out + assert exit_code == 1 + assert settings_loaded is False + assert "configuration could not be read" in output + assert "output could not be written" not in output + + +def test_main_reports_malformed_json_line_before_loading_settings( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """Malformed JSON must report its line without exposing the document.""" + settings_loaded = False + config_path = tmp_path / "malformed.json" + config_path.write_text('{\n "repositories": [\n}\n', encoding="utf-8") + + def unexpected_settings(cls: type) -> None: + nonlocal settings_loaded + settings_loaded = True + raise AssertionError("Settings.from_env must not be called") + + monkeypatch.setattr( + extract_repositories.Settings, + "from_env", + classmethod(unexpected_settings), + ) + + exit_code = extract_repositories.main( + [ + "--config", + str(config_path), + "--output-root", + str(tmp_path / "data"), + ] + ) + + output = capsys.readouterr().out + assert exit_code == 1 + assert settings_loaded is False + assert "invalid JSON at line 3" in output + assert '"repositories"' not in output + + +def test_main_reports_invalid_utf8_before_loading_settings( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """Invalid UTF-8 must be reported safely before credentials are loaded.""" + settings_loaded = False + config_path = tmp_path / "invalid-utf8.json" + config_path.write_bytes(b'\xff\xfe{"binary-content": "must-not-leak"}') + + def unexpected_settings(cls: type) -> None: + nonlocal settings_loaded + settings_loaded = True + raise AssertionError("Settings.from_env must not be called") + + monkeypatch.setattr( + extract_repositories.Settings, + "from_env", + classmethod(unexpected_settings), + ) + + exit_code = extract_repositories.main( + [ + "--config", + str(config_path), + "--output-root", + str(tmp_path / "data"), + ] + ) + + output = capsys.readouterr().out + assert exit_code == 1 + assert settings_loaded is False + assert "valid UTF-8" in output + assert "binary-content" not in output + assert "must-not-leak" not in output + assert "output could not be written" not in output diff --git a/tests/test_repository_batch.py b/tests/test_repository_batch.py new file mode 100644 index 0000000..17cc3fb --- /dev/null +++ b/tests/test_repository_batch.py @@ -0,0 +1,506 @@ +"""Tests for configurable repository batch extraction and local persistence.""" + +from __future__ import annotations + +import json +import os +from collections.abc import Iterator +from datetime import UTC, datetime +from pathlib import Path +from unittest.mock import Mock + +import pytest + +from ingestion.github.exceptions import GitHubNotFoundError +from ingestion.github.repository_batch import ( + GitHubRepositoryBatch, + RepositoryBatchConfigurationError, + load_repository_config, +) +from ingestion.github.repository_metadata import RepositoryMetadata +from ingestion.github.repository_service import ( + GitHubRepositoryService, + RepositoryExtractionResult, +) + + +def build_payload( + *, + owner: str = "apache", + repository: str = "airflow", + repository_id: int = 1, +) -> dict[str, object]: + """Build one representative GitHub repository payload.""" + return { + "id": repository_id, + "name": repository, + "full_name": f"{owner}/{repository}", + "owner": {"login": owner}, + "description": "Example repository", + "visibility": "public", + "default_branch": "main", + "language": "Python", + "fork": False, + "archived": False, + "disabled": False, + "created_at": "2020-01-01T00:00:00Z", + "updated_at": "2026-07-23T12:00:00Z", + "pushed_at": "2026-07-23T11:00:00Z", + "stargazers_count": 10, + "forks_count": 2, + "open_issues_count": 1, + "subscribers_count": 3, + "size": 100, + "html_url": f"https://github.com/{owner}/{repository}", + "custom_source_field": {"preserved": True}, + } + + +def build_result( + extracted_at: datetime, + *, + owner: str = "apache", + repository: str = "airflow", + repository_id: int = 1, +) -> RepositoryExtractionResult: + """Build one service result with its untouched source payload.""" + payload = build_payload( + owner=owner, + repository=repository, + repository_id=repository_id, + ) + return RepositoryExtractionResult( + endpoint=f"/repos/{owner}/{repository}", + payload=payload, + metadata=RepositoryMetadata.from_github_payload( + payload, + extracted_at=extracted_at, + ), + ) + + +def write_config(path: Path, repositories: object) -> None: + """Write a JSON configuration fixture.""" + path.write_text( + json.dumps({"repositories": repositories}), + encoding="utf-8", + ) + + +def sequence_clock(values: list[datetime]): + """Return a deterministic clock over the provided values.""" + iterator: Iterator[datetime] = iter(values) + return lambda: next(iterator) + + +def test_load_repository_config_accepts_multiple_repositories(tmp_path: Path) -> None: + """A valid configuration must preserve every requested repository.""" + config_path = tmp_path / "repositories.json" + write_config( + config_path, + [ + {"owner": "apache", "name": "airflow"}, + {"owner": "pandas-dev", "name": "pandas"}, + ], + ) + + config = load_repository_config(config_path) + + assert [(item.owner, item.name) for item in config.repositories] == [ + ("apache", "airflow"), + ("pandas-dev", "pandas"), + ] + + +@pytest.mark.parametrize( + ("document", "message"), + [ + ([], "root.*object"), + ({}, "repositories"), + ({"repositories": []}, "non-empty list"), + ({"repositories": ["apache/airflow"]}, "item 1.*object"), + ({"repositories": [{"name": "airflow"}]}, "owner"), + ({"repositories": [{"owner": "apache", "name": ""}]}, "name"), + ({"repositories": [{"owner": "apache/team", "name": "airflow"}]}, "separator"), + ], +) +def test_load_repository_config_rejects_invalid_documents( + tmp_path: Path, + document: object, + message: str, +) -> None: + """Invalid configuration shapes must produce clear validation errors.""" + config_path = tmp_path / "repositories.json" + config_path.write_text(json.dumps(document), encoding="utf-8") + + with pytest.raises(RepositoryBatchConfigurationError, match=message): + load_repository_config(config_path) + + +def test_load_repository_config_rejects_case_insensitive_duplicates( + tmp_path: Path, +) -> None: + """GitHub repository identifiers must be unique regardless of case.""" + config_path = tmp_path / "repositories.json" + write_config( + config_path, + [ + {"owner": "Apache", "name": "Airflow"}, + {"owner": "apache", "name": "airflow"}, + ], + ) + + with pytest.raises(RepositoryBatchConfigurationError, match="duplicate"): + load_repository_config(config_path) + + +def test_load_repository_config_wraps_read_errors(tmp_path: Path) -> None: + """Unreadable configuration must produce a safe configuration error.""" + missing_path = tmp_path / "missing.json" + + with pytest.raises( + RepositoryBatchConfigurationError, + match="could not be read", + ) as exception_info: + load_repository_config(missing_path) + + assert isinstance(exception_info.value.__cause__, OSError) + + +def test_load_repository_config_rejects_invalid_utf8(tmp_path: Path) -> None: + """Configuration bytes must contain valid UTF-8 without leaking content.""" + config_path = tmp_path / "invalid-utf8.json" + config_path.write_bytes(b'\xff\xfe{"binary-content": "must-not-leak"}') + + with pytest.raises( + RepositoryBatchConfigurationError, + match="valid UTF-8", + ) as exception_info: + load_repository_config(config_path) + + assert isinstance(exception_info.value.__cause__, UnicodeError) + + +def test_batch_persists_raw_normalized_and_manifest_with_shared_timestamp( + tmp_path: Path, +) -> None: + """Successful records must share one timestamp and preserve source data.""" + started_at = datetime(2026, 7, 23, 12, 0, 0, tzinfo=UTC) + extracted_at = datetime(2026, 7, 23, 12, 0, 1, 123456, tzinfo=UTC) + completed_at = datetime(2026, 7, 23, 12, 0, 5, tzinfo=UTC) + + config_path = tmp_path / "repositories.json" + write_config( + config_path, + [ + {"owner": "apache", "name": "airflow"}, + {"owner": "pandas-dev", "name": "pandas"}, + ], + ) + config = load_repository_config(config_path) + + service = Mock(spec=GitHubRepositoryService) + service.extract_repository_with_payload.side_effect = [ + build_result(extracted_at), + build_result( + extracted_at, + owner="pandas-dev", + repository="pandas", + repository_id=2, + ), + ] + batch = GitHubRepositoryBatch( + service, + clock=sequence_clock([started_at, extracted_at, completed_at]), + ) + + result = batch.run(config, output_root=tmp_path / "data") + + assert result.status == "success" + assert result.run_id == "20260723T120001123456Z" + assert result.manifest_path.exists() + + manifest = json.loads(result.manifest_path.read_text(encoding="utf-8")) + assert manifest["started_at"] == started_at.isoformat() + assert manifest["extracted_at"] == extracted_at.isoformat() + assert manifest["completed_at"] == completed_at.isoformat() + assert manifest["repositories_requested"] == 2 + assert manifest["repositories_succeeded"] == 2 + assert manifest["repositories_failed"] == 0 + assert manifest["raw_output"].startswith("raw/") + assert manifest["normalized_output"].startswith("normalized/") + assert "\\" not in manifest["raw_output"] + + raw_path = result.output_root / manifest["raw_output"] + normalized_path = result.output_root / manifest["normalized_output"] + raw_records = [ + json.loads(line) for line in raw_path.read_text(encoding="utf-8").splitlines() + ] + normalized_records = [ + json.loads(line) + for line in normalized_path.read_text(encoding="utf-8").splitlines() + ] + + assert len(raw_records) == len(normalized_records) == 2 + assert raw_records[0]["payload"] == build_payload() + assert raw_records[0]["run_id"] == result.run_id + assert raw_records[0]["extracted_at"] == extracted_at.isoformat() + assert normalized_records[0]["run_id"] == result.run_id + assert normalized_records[0]["created_at"] == "2020-01-01T00:00:00+00:00" + assert normalized_records[0]["extracted_at"] == extracted_at.isoformat() + + for call in service.extract_repository_with_payload.call_args_list: + assert call.kwargs["extracted_at"] == extracted_at + + +def test_batch_continues_after_failure_and_reports_partial_success( + tmp_path: Path, +) -> None: + """One repository failure must not stop subsequent repositories.""" + times = [datetime(2026, 7, 23, 13, 0, second, tzinfo=UTC) for second in (0, 1, 2)] + config_path = tmp_path / "repositories.json" + write_config( + config_path, + [ + {"owner": "missing", "name": "repository"}, + {"owner": "apache", "name": "airflow"}, + ], + ) + config = load_repository_config(config_path) + service = Mock(spec=GitHubRepositoryService) + service.extract_repository_with_payload.side_effect = [ + GitHubNotFoundError("sensitive source response"), + build_result(times[1]), + ] + + result = GitHubRepositoryBatch( + service, + clock=sequence_clock(times), + ).run(config, output_root=tmp_path / "data") + + manifest = json.loads(result.manifest_path.read_text(encoding="utf-8")) + assert result.status == "partial_success" + assert manifest["repositories_succeeded"] == 1 + assert manifest["repositories_failed"] == 1 + assert manifest["errors"] == [ + { + "owner": "missing", + "repository": "repository", + "error_type": "GitHubNotFoundError", + "message": "GitHub repository was not found.", + } + ] + assert service.extract_repository_with_payload.call_count == 2 + + +def test_batch_continues_without_orphan_raw_after_serialization_failure( + tmp_path: Path, +) -> None: + """Raw and normalized records must be committed per repository as one unit.""" + times = [datetime(2026, 7, 23, 13, 30, second, tzinfo=UTC) for second in (0, 1, 2)] + config_path = tmp_path / "repositories.json" + write_config( + config_path, + [ + {"owner": "broken", "name": "repository"}, + {"owner": "apache", "name": "airflow"}, + ], + ) + config = load_repository_config(config_path) + broken_metadata = Mock(spec=RepositoryMetadata) + broken_metadata.to_record.side_effect = ValueError( + "Authorization: Bearer simulated-token" + ) + service = Mock(spec=GitHubRepositoryService) + service.extract_repository_with_payload.side_effect = [ + RepositoryExtractionResult( + endpoint="/repos/broken/repository", + payload={"id": "raw-must-not-be-written"}, + metadata=broken_metadata, + ), + build_result(times[1]), + ] + + result = GitHubRepositoryBatch( + service, + clock=sequence_clock(times), + ).run(config, output_root=tmp_path / "data") + + manifest = json.loads(result.manifest_path.read_text(encoding="utf-8")) + raw_path = result.output_root / manifest["raw_output"] + normalized_path = result.output_root / manifest["normalized_output"] + raw_records = [ + json.loads(line) for line in raw_path.read_text(encoding="utf-8").splitlines() + ] + normalized_records = [ + json.loads(line) + for line in normalized_path.read_text(encoding="utf-8").splitlines() + ] + + assert result.status == "partial_success" + assert service.extract_repository_with_payload.call_count == 2 + assert len(raw_records) == len(normalized_records) == 1 + assert raw_records[0]["owner"] == "apache" + assert normalized_records[0]["owner_login"] == "apache" + assert manifest["repositories_succeeded"] == 1 + assert manifest["repositories_failed"] == 1 + assert manifest["errors"][0] == { + "owner": "broken", + "repository": "repository", + "error_type": "UnexpectedRepositoryError", + "message": "Unexpected repository extraction error.", + } + assert "simulated-token" not in result.manifest_path.read_text(encoding="utf-8") + + +def test_batch_reports_failed_when_every_repository_fails(tmp_path: Path) -> None: + """A run without successful repositories must be marked failed.""" + times = [datetime(2026, 7, 23, 14, 0, second, tzinfo=UTC) for second in (0, 1, 2)] + config_path = tmp_path / "repositories.json" + write_config(config_path, [{"owner": "missing", "name": "one"}]) + config = load_repository_config(config_path) + service = Mock(spec=GitHubRepositoryService) + service.extract_repository_with_payload.side_effect = GitHubNotFoundError( + "not found" + ) + + result = GitHubRepositoryBatch( + service, + clock=sequence_clock(times), + ).run(config, output_root=tmp_path / "data") + + manifest = json.loads(result.manifest_path.read_text(encoding="utf-8")) + assert result.status == "failed" + assert manifest["repositories_succeeded"] == 0 + assert manifest["repositories_failed"] == 1 + + +def test_unexpected_repository_error_does_not_persist_secrets(tmp_path: Path) -> None: + """Unexpected exception content must never be copied to the manifest.""" + secret = "Authorization: Bearer simulated-token" + times = [datetime(2026, 7, 23, 15, 0, second, tzinfo=UTC) for second in (0, 1, 2)] + config_path = tmp_path / "repositories.json" + write_config(config_path, [{"owner": "apache", "name": "airflow"}]) + config = load_repository_config(config_path) + service = Mock(spec=GitHubRepositoryService) + service.extract_repository_with_payload.side_effect = RuntimeError(secret) + + result = GitHubRepositoryBatch( + service, + clock=sequence_clock(times), + ).run(config, output_root=tmp_path / "data") + + manifest_text = result.manifest_path.read_text(encoding="utf-8") + assert "Authorization" not in manifest_text + assert "simulated-token" not in manifest_text + persisted_error = json.loads(manifest_text)["errors"][0] + assert persisted_error["error_type"] == "UnexpectedRepositoryError" + assert persisted_error["message"] == "Unexpected repository extraction error." + + +def test_atomic_writer_publishes_manifest_last_and_uses_local_temporaries( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """All local temporaries must close before outputs are published in order.""" + times = [datetime(2026, 7, 23, 16, 0, second, tzinfo=UTC) for second in (0, 1, 2)] + config_path = tmp_path / "repositories.json" + write_config(config_path, [{"owner": "apache", "name": "airflow"}]) + config = load_repository_config(config_path) + service = Mock(spec=GitHubRepositoryService) + service.extract_repository_with_payload.return_value = build_result(times[1]) + replacements: list[tuple[Path, Path]] = [] + events: list[str] = [] + real_replace = os.replace + clock_values: Iterator[datetime] = iter(times) + clock_reads = 0 + + def event_clock() -> datetime: + nonlocal clock_reads + clock_reads += 1 + if clock_reads == 3: + events.append("completed_at") + return next(clock_values) + + def recording_replace(source: str | Path, destination: str | Path) -> None: + source_path = Path(source) + destination_path = Path(destination) + assert source_path.parent == destination_path.parent + replacements.append((source_path, destination_path)) + events.append(destination_path.parts[-6]) + real_replace(source_path, destination_path) + + monkeypatch.setattr( + "ingestion.github.repository_batch.os.replace", recording_replace + ) + + result = GitHubRepositoryBatch( + service, + clock=event_clock, + ).run(config, output_root=tmp_path / "data") + + assert events == ["raw", "normalized", "completed_at", "manifests"] + assert [destination.parts[-6] for _, destination in replacements] == [ + "raw", + "normalized", + "manifests", + ] + assert replacements[-1][1] == result.manifest_path + assert not list(result.output_root.rglob("*.tmp")) + + +def test_atomic_writer_rejects_existing_run_outputs(tmp_path: Path) -> None: + """A repeated run identifier must never overwrite prior final files.""" + times = [datetime(2026, 7, 23, 17, 0, second, tzinfo=UTC) for second in (0, 1, 2)] + config_path = tmp_path / "repositories.json" + write_config(config_path, [{"owner": "apache", "name": "airflow"}]) + config = load_repository_config(config_path) + service = Mock(spec=GitHubRepositoryService) + service.extract_repository_with_payload.return_value = build_result(times[1]) + + GitHubRepositoryBatch( + service, + clock=sequence_clock(times), + ).run(config, output_root=tmp_path / "data") + + repeated_clock = sequence_clock(times[:2]) + with pytest.raises(FileExistsError, match="already exists"): + GitHubRepositoryBatch(service, clock=repeated_clock).run( + config, + output_root=tmp_path / "data", + ) + + +def test_atomic_writer_cleans_only_remaining_temporaries_on_publish_failure( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A publication failure must remove run temporaries without deleting finals.""" + times = [datetime(2026, 7, 23, 18, 0, second, tzinfo=UTC) for second in (0, 1, 2)] + config_path = tmp_path / "repositories.json" + write_config(config_path, [{"owner": "apache", "name": "airflow"}]) + config = load_repository_config(config_path) + service = Mock(spec=GitHubRepositoryService) + service.extract_repository_with_payload.return_value = build_result(times[1]) + real_replace = os.replace + replacement_count = 0 + + def failing_replace(source: str | Path, destination: str | Path) -> None: + nonlocal replacement_count + replacement_count += 1 + if replacement_count == 2: + raise OSError("simulated publication failure") + real_replace(source, destination) + + monkeypatch.setattr("ingestion.github.repository_batch.os.replace", failing_replace) + + with pytest.raises(OSError, match="publication failure"): + GitHubRepositoryBatch( + service, + clock=sequence_clock(times), + ).run(config, output_root=tmp_path / "data") + + output_root = tmp_path / "data" + assert len(list(output_root.rglob("repositories.jsonl"))) == 1 + assert not list(output_root.rglob("manifest.json")) + assert not list(output_root.rglob("*.tmp")) diff --git a/tests/test_repository_service.py b/tests/test_repository_service.py index 0d776b0..1d26e8d 100644 --- a/tests/test_repository_service.py +++ b/tests/test_repository_service.py @@ -72,6 +72,24 @@ def test_extract_repository_returns_normalized_metadata() -> None: assert metadata.extracted_at == extracted_at +def test_extract_repository_with_payload_returns_structured_result() -> None: + """The extended service must expose raw and normalized data from one request.""" + payload = build_repository_payload() + client = Mock(spec=GitHubClient) + client.get.return_value = payload + service = GitHubRepositoryService(client) + + result = service.extract_repository_with_payload( + owner="apache", + repository="airflow", + ) + + client.get.assert_called_once_with("/repos/apache/airflow") + assert result.endpoint == "/repos/apache/airflow" + assert result.payload is payload + assert result.metadata.full_name == "apache/airflow" + + def test_extract_repository_rejects_non_object_response() -> None: """The repository endpoint must return a JSON object.""" client = Mock(spec=GitHubClient) From 17b923343e6e5fab8163522afd888c7101c8970e Mon Sep 17 00:00:00 2001 From: Daniel Burbano Date: Thu, 23 Jul 2026 20:10:24 -0500 Subject: [PATCH 5/8] chore: remove obsolete GitHub client backup --- ingestion/github/client.py.bak | 221 --------------------------------- 1 file changed, 221 deletions(-) delete mode 100644 ingestion/github/client.py.bak diff --git a/ingestion/github/client.py.bak b/ingestion/github/client.py.bak deleted file mode 100644 index ece9d57..0000000 --- a/ingestion/github/client.py.bak +++ /dev/null @@ -1,221 +0,0 @@ -"""Reusable HTTP client for the GitHub REST API.""" - -from __future__ import annotations - -from typing import Any - -import requests -from requests import Response, Session -from requests.exceptions import ( - ConnectionError as RequestsConnectionError, -) -from requests.exceptions import ( - HTTPError, - JSONDecodeError, - Timeout, -) - -from ingestion.github.exceptions import ( - GitHubAPIError, - GitHubAuthenticationError, - GitHubNotFoundError, - GitHubRateLimitError, -) - -type JSONData = dict[str, Any] | list[Any] - - -class GitHubClient: - """Client responsible for sending authenticated requests to GitHub.""" - - def __init__( - self, - token: str, - base_url: str = "https://api.github.com", - api_version: str = "2026-03-10", - timeout_seconds: float = 30.0, - session: Session | None = None, - ) -> None: - """Initialize the GitHub API client.""" - clean_token = token.strip() - - if not clean_token: - raise ValueError("A GitHub token is required.") - - if timeout_seconds <= 0: - raise ValueError("timeout_seconds must be greater than zero.") - - self._base_url = base_url.rstrip("/") - self._timeout_seconds = timeout_seconds - self._session = session or requests.Session() - - self._session.headers.update( - { - "Accept": "application/vnd.github+json", - "Authorization": f"Bearer {clean_token}", - "X-GitHub-Api-Version": api_version, - "User-Agent": "devflow-engineering-analytics", - } - ) - - def get( - self, - endpoint: str, - params: dict[str, Any] | None = None, - ) -> JSONData: - """Send one GET request and return its decoded JSON response.""" - url = self._build_url(endpoint) - response = self._send_get(url=url, params=params) - - self._raise_for_status(response) - - return self._decode_json(response=response, request_url=url) - - def get_all_pages( - self, - endpoint: str, - params: dict[str, Any] | None = None, - max_pages: int | None = None, - ) -> list[Any]: - """Retrieve and combine every page from a collection endpoint. - - The optional max_pages argument limits the number of pages. It is - useful during local development to avoid downloading large datasets. - """ - if max_pages is not None and max_pages <= 0: - raise ValueError("max_pages must be greater than zero.") - - url = self._build_url(endpoint) - - request_params = dict(params or {}) - request_params.setdefault("per_page", 100) - - all_items: list[Any] = [] - page_count = 0 - - while url: - response = self._send_get( - url=url, - params=request_params, - ) - - self._raise_for_status(response) - - payload = self._decode_json( - response=response, - request_url=url, - ) - - if not isinstance(payload, list): - raise GitHubAPIError( - "A paginated GitHub endpoint must return a JSON list." - ) - - all_items.extend(payload) - page_count += 1 - - if max_pages is not None and page_count >= max_pages: - break - - next_url = response.links.get("next", {}).get("url") - - if isinstance(next_url, str): - url = next_url - else: - url = "" - - # The next URL supplied by GitHub already contains its query - # parameters, so the original parameters must not be sent again. - request_params = None - - return all_items - - def _send_get( - self, - url: str, - params: dict[str, Any] | None, - ) -> Response: - """Send a GET request while handling network-level failures.""" - try: - return self._session.get( - url, - params=params, - timeout=self._timeout_seconds, - ) - except Timeout as exc: - raise GitHubAPIError( - f"The GitHub request timed out after {self._timeout_seconds} seconds." - ) from exc - except RequestsConnectionError as exc: - raise GitHubAPIError("The connection to the GitHub API failed.") from exc - - def _decode_json( - self, - response: Response, - request_url: str, - ) -> JSONData: - """Decode and validate a GitHub JSON response.""" - try: - payload = response.json() - except JSONDecodeError as exc: - raise GitHubAPIError( - f"GitHub returned invalid JSON for {request_url}." - ) from exc - - if not isinstance(payload, (dict, list)): - raise GitHubAPIError( - f"GitHub returned an unsupported JSON structure for {request_url}." - ) - - return payload - - def _raise_for_status(self, response: Response) -> None: - """Convert relevant HTTP status codes into domain exceptions.""" - status_code = response.status_code - request_url = response.url - - if status_code == 401: - raise GitHubAuthenticationError( - "GitHub rejected the token. Verify GITHUB_TOKEN." - ) - - if status_code == 404: - raise GitHubNotFoundError( - f"The requested GitHub resource was not found: {request_url}" - ) - - if status_code in {403, 429}: - remaining = response.headers.get("x-ratelimit-remaining") - reset_at = response.headers.get("x-ratelimit-reset") - retry_after = response.headers.get("retry-after") - - if remaining == "0" or retry_after is not None: - raise GitHubRateLimitError( - "GitHub API rate limit reached. " - f"Remaining={remaining}, " - f"reset_at={reset_at}, " - f"retry_after={retry_after}." - ) - - raise GitHubAPIError( - f"GitHub rejected the request with status {status_code}." - ) - - try: - response.raise_for_status() - except HTTPError as exc: - raise GitHubAPIError( - f"GitHub request failed with status {status_code}: {request_url}" - ) from exc - - def _build_url(self, endpoint: str) -> str: - """Build a complete GitHub API URL from a relative endpoint.""" - clean_endpoint = endpoint.strip() - - if not clean_endpoint: - raise ValueError("The endpoint cannot be empty.") - - if not clean_endpoint.startswith("/"): - clean_endpoint = f"/{clean_endpoint}" - - return f"{self._base_url}{clean_endpoint}" From a99e45c9a2bd626ec1b9574a58b695a55b5c8a60 Mon Sep 17 00:00:00 2001 From: Daniel Burbano Date: Fri, 18 Sep 2026 16:24:18 -0500 Subject: [PATCH 6/8] feat: add reproducible repository pipeline --- ingestion/github/repository_batch.py | 26 +++ ingestion/github/repository_metadata.py | 18 +- orchestration/__init__.py | 17 ++ orchestration/pipeline.py | 271 ++++++++++++++++++++++++ scripts/run_devflow.py | 45 ++++ tests/test_pipeline.py | 253 ++++++++++++++++++++++ tests/test_repository_batch.py | 9 + tests/test_repository_metadata.py | 14 ++ tests/test_run_devflow.py | 104 +++++++++ 9 files changed, 756 insertions(+), 1 deletion(-) create mode 100644 orchestration/__init__.py create mode 100644 orchestration/pipeline.py create mode 100644 scripts/run_devflow.py create mode 100644 tests/test_pipeline.py create mode 100644 tests/test_run_devflow.py diff --git a/ingestion/github/repository_batch.py b/ingestion/github/repository_batch.py index 0323715..7cf6d40 100644 --- a/ingestion/github/repository_batch.py +++ b/ingestion/github/repository_batch.py @@ -167,6 +167,16 @@ def run( ) _reject_existing_outputs(paths) + logger.info( + "GitHub repository batch started.", + extra={ + "operation": "github_repository_batch", + "run_id": run_id, + "status": "started", + "repositories_requested": len(config.repositories), + }, + ) + raw_records: list[dict[str, Any]] = [] normalized_records: list[dict[str, Any]] = [] errors: list[dict[str, str]] = [] @@ -197,6 +207,7 @@ def run( "GitHub repository batch item failed.", extra={ "operation": "github_repository_batch_item", + "run_id": run_id, "owner": repository.owner, "repository": repository.name, "error_type": error["error_type"], @@ -218,6 +229,8 @@ def run( "repositories_requested": len(config.repositories), "repositories_succeeded": succeeded, "repositories_failed": failed, + "raw_records_written": succeeded, + "normalized_records_written": succeeded, "raw_output": paths.raw.relative_to(output_root).as_posix(), "normalized_output": paths.normalized.relative_to(output_root).as_posix(), "errors": errors, @@ -232,6 +245,19 @@ def run( clock=self._clock, ) + logger.info( + "GitHub repository batch completed.", + extra={ + "operation": "github_repository_batch", + "run_id": run_id, + "status": status, + "repositories_succeeded": succeeded, + "repositories_failed": failed, + "raw_records_written": succeeded, + "normalized_records_written": succeeded, + }, + ) + return RepositoryBatchResult( run_id=run_id, status=status, diff --git a/ingestion/github/repository_metadata.py b/ingestion/github/repository_metadata.py index 6644d04..ca0a820 100644 --- a/ingestion/github/repository_metadata.py +++ b/ingestion/github/repository_metadata.py @@ -77,7 +77,7 @@ def from_github_payload( ) return cls( - repository_id=_require_non_negative_integer( + repository_id=_require_positive_integer( payload, field_name="id", ), @@ -263,6 +263,22 @@ def _require_non_negative_integer( return value +def _require_positive_integer( + payload: Mapping[str, Any], + *, + field_name: str, +) -> int: + """Return a required strictly positive integer field.""" + value = _require_non_negative_integer(payload, field_name=field_name) + + if value == 0: + raise RepositoryMetadataError( + f"Field '{field_name}' must be greater than zero." + ) + + return value + + def _require_datetime( payload: Mapping[str, Any], *, diff --git a/orchestration/__init__.py b/orchestration/__init__.py new file mode 100644 index 0000000..109d24d --- /dev/null +++ b/orchestration/__init__.py @@ -0,0 +1,17 @@ +"""Local orchestration for DevFlow Intelligence pipelines.""" + +from orchestration.pipeline import ( + PipelineResult, + PipelineStageResult, + PipelineStageStatus, + PipelineStatus, + run_pipeline, +) + +__all__ = [ + "PipelineResult", + "PipelineStageResult", + "PipelineStageStatus", + "PipelineStatus", + "run_pipeline", +] diff --git a/orchestration/pipeline.py b/orchestration/pipeline.py new file mode 100644 index 0000000..0653680 --- /dev/null +++ b/orchestration/pipeline.py @@ -0,0 +1,271 @@ +"""Dependency composition and local orchestration for repository ingestion.""" + +from __future__ import annotations + +from collections.abc import Callable, Iterator +from dataclasses import dataclass +from datetime import UTC, datetime +from enum import StrEnum +from pathlib import Path +from typing import Any + +from ingestion.common.config import ConfigurationError, Settings +from ingestion.common.logging import configure_logging +from ingestion.github.client import GitHubClient +from ingestion.github.repository_batch import ( + GitHubRepositoryBatch, + RepositoryBatchConfigurationError, + RepositoryBatchResult, + load_repository_config, +) +from ingestion.github.repository_service import GitHubRepositoryService + +type Clock = Callable[[], datetime] + + +class PipelineStatus(StrEnum): + """Terminal status of one pipeline execution.""" + + SUCCESS = "success" + PARTIAL_SUCCESS = "partial_success" + FAILED = "failed" + + +class PipelineStageStatus(StrEnum): + """Status of one pipeline stage.""" + + SUCCESS = "success" + FAILED = "failed" + SKIPPED = "skipped" + + +@dataclass(frozen=True, slots=True) +class PipelineStageResult: + """Automation-safe result for one pipeline stage.""" + + name: str + status: PipelineStageStatus + + def to_record(self) -> dict[str, str]: + """Return a JSON-serializable representation.""" + return {"name": self.name, "status": self.status.value} + + +@dataclass(frozen=True, slots=True) +class PipelineResult: + """Stable result returned by every pipeline execution.""" + + run_id: str + started_at: datetime + extracted_at: datetime + status: PipelineStatus + repositories_succeeded: int + repositories_failed: int + manifest_path: Path | None + stages: tuple[PipelineStageResult, ...] + error_type: str | None = None + error_message: str | None = None + + @property + def exit_code(self) -> int: + """Map terminal status to a process exit code.""" + return { + PipelineStatus.SUCCESS: 0, + PipelineStatus.PARTIAL_SUCCESS: 2, + PipelineStatus.FAILED: 1, + }[self.status] + + def to_record(self) -> dict[str, Any]: + """Return a JSON-serializable, secret-safe result.""" + return { + "run_id": self.run_id, + "started_at": self.started_at.isoformat(), + "extracted_at": self.extracted_at.isoformat(), + "status": self.status.value, + "exit_code": self.exit_code, + "repositories_succeeded": self.repositories_succeeded, + "repositories_failed": self.repositories_failed, + "manifest_path": ( + str(self.manifest_path) if self.manifest_path is not None else None + ), + "stages": [stage.to_record() for stage in self.stages], + "error_type": self.error_type, + "error_message": self.error_message, + } + + +def run_pipeline( + *, + config_path: Path, + output_root: Path, + clock: Clock | None = None, +) -> PipelineResult: + """Run configuration through GitHub extraction and local persistence.""" + pipeline_clock = clock or _utc_now + started_at = _read_utc_clock(pipeline_clock, field_name="started_at") + extracted_at = _read_utc_clock(pipeline_clock, field_name="extracted_at") + run_id = extracted_at.strftime("%Y%m%dT%H%M%S%fZ") + stages: list[PipelineStageResult] = [] + + try: + repository_config = load_repository_config(Path(config_path)) + settings = Settings.from_env() + configure_logging(settings.log_level) + except (ConfigurationError, RepositoryBatchConfigurationError) as exc: + stages.append(_stage("configuration", PipelineStageStatus.FAILED)) + stages.extend(_skipped_stages("client", "service", "batch")) + return _failed_result( + run_id=run_id, + started_at=started_at, + extracted_at=extracted_at, + stages=stages, + error=exc, + message=str(exc), + ) + + stages.append(_stage("configuration", PipelineStageStatus.SUCCESS)) + + try: + client = GitHubClient( + token=settings.github_token, + base_url=settings.github_api_base_url, + api_version=settings.github_api_version, + timeout_seconds=settings.github_timeout_seconds, + max_retries=settings.github_max_retries, + backoff_seconds=settings.github_backoff_seconds, + ) + except (TypeError, ValueError) as exc: + stages.append(_stage("client", PipelineStageStatus.FAILED)) + stages.extend(_skipped_stages("service", "batch")) + return _failed_result( + run_id=run_id, + started_at=started_at, + extracted_at=extracted_at, + stages=stages, + error=exc, + message="GitHub client configuration is invalid.", + ) + + stages.append(_stage("client", PipelineStageStatus.SUCCESS)) + service = GitHubRepositoryService(client) + stages.append(_stage("service", PipelineStageStatus.SUCCESS)) + + batch_clock = _BatchClock( + started_at=started_at, + extracted_at=extracted_at, + completion_clock=pipeline_clock, + ) + + try: + batch_result = GitHubRepositoryBatch(service, clock=batch_clock).run( + repository_config, + output_root=Path(output_root), + ) + except OSError as exc: + stages.append(_stage("batch", PipelineStageStatus.FAILED)) + return _failed_result( + run_id=run_id, + started_at=started_at, + extracted_at=extracted_at, + stages=stages, + error=exc, + message="Pipeline outputs could not be published.", + ) + + stages.append(_stage("batch", PipelineStageStatus.SUCCESS)) + return _successful_result( + batch_result=batch_result, + started_at=started_at, + extracted_at=extracted_at, + stages=stages, + ) + + +class _BatchClock: + """Give the batch the pipeline timestamps, then a live completion time.""" + + def __init__( + self, + *, + started_at: datetime, + extracted_at: datetime, + completion_clock: Clock, + ) -> None: + self._initial_values: Iterator[datetime] = iter((started_at, extracted_at)) + self._completion_clock = completion_clock + + def __call__(self) -> datetime: + """Return shared context values before delegating completion time.""" + try: + return next(self._initial_values) + except StopIteration: + return self._completion_clock() + + +def _utc_now() -> datetime: + """Return the current UTC time.""" + return datetime.now(UTC) + + +def _read_utc_clock(clock: Clock, *, field_name: str) -> datetime: + """Read one timezone-aware clock value and normalize it to UTC.""" + value = clock() + if not isinstance(value, datetime): + raise ValueError(f"{field_name} clock value must be a datetime.") + if value.tzinfo is None or value.utcoffset() is None: + raise ValueError(f"{field_name} clock value must include timezone information.") + return value.astimezone(UTC) + + +def _stage(name: str, status: PipelineStageStatus) -> PipelineStageResult: + """Build one stage result.""" + return PipelineStageResult(name=name, status=status) + + +def _skipped_stages(*names: str) -> list[PipelineStageResult]: + """Build skipped results for stages that could not run.""" + return [_stage(name, PipelineStageStatus.SKIPPED) for name in names] + + +def _failed_result( + *, + run_id: str, + started_at: datetime, + extracted_at: datetime, + stages: list[PipelineStageResult], + error: Exception, + message: str, +) -> PipelineResult: + """Build a failed result without copying unsafe exception details.""" + return PipelineResult( + run_id=run_id, + started_at=started_at, + extracted_at=extracted_at, + status=PipelineStatus.FAILED, + repositories_succeeded=0, + repositories_failed=0, + manifest_path=None, + stages=tuple(stages), + error_type=type(error).__name__, + error_message=message, + ) + + +def _successful_result( + *, + batch_result: RepositoryBatchResult, + started_at: datetime, + extracted_at: datetime, + stages: list[PipelineStageResult], +) -> PipelineResult: + """Adapt a repository batch result to the public pipeline contract.""" + return PipelineResult( + run_id=batch_result.run_id, + started_at=started_at, + extracted_at=extracted_at, + status=PipelineStatus(batch_result.status), + repositories_succeeded=batch_result.repositories_succeeded, + repositories_failed=batch_result.repositories_failed, + manifest_path=batch_result.manifest_path, + stages=tuple(stages), + ) diff --git a/scripts/run_devflow.py b/scripts/run_devflow.py new file mode 100644 index 0000000..1a1f662 --- /dev/null +++ b/scripts/run_devflow.py @@ -0,0 +1,45 @@ +"""Run the DevFlow repository ingestion pipeline.""" + +from __future__ import annotations + +import argparse +import json +from collections.abc import Sequence +from pathlib import Path + +from orchestration.pipeline import run_pipeline + + +def build_parser() -> argparse.ArgumentParser: + """Build the local pipeline command-line parser.""" + parser = argparse.ArgumentParser( + description="Run GitHub repository ingestion through local persistence." + ) + parser.add_argument( + "--config", + required=True, + type=Path, + help="Path to the repository configuration JSON file.", + ) + parser.add_argument( + "--output-root", + required=True, + type=Path, + help="Root directory for raw, normalized and manifest outputs.", + ) + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + """Execute the pipeline and print its automation-safe JSON result.""" + args = build_parser().parse_args(argv) + result = run_pipeline( + config_path=args.config, + output_root=args.output_root, + ) + print(json.dumps(result.to_record(), ensure_ascii=False)) + return result.exit_code + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py new file mode 100644 index 0000000..d3fd3ae --- /dev/null +++ b/tests/test_pipeline.py @@ -0,0 +1,253 @@ +"""Tests for local DevFlow pipeline orchestration.""" + +from __future__ import annotations + +from datetime import UTC, datetime +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from ingestion.github.repository_batch import ( + RepositoryBatchConfigurationError, + RepositoryBatchResult, +) +from orchestration import pipeline + + +def sequence_clock(values: list[datetime]): + """Return a deterministic clock over the provided values.""" + iterator = iter(values) + return lambda: next(iterator) + + +def test_pipeline_composes_dependencies_with_one_shared_run_context( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """Every batch layer must receive the run timestamp owned by orchestration.""" + started_at = datetime(2026, 9, 18, 10, 0, 0, tzinfo=UTC) + extracted_at = datetime(2026, 9, 18, 10, 0, 1, 123456, tzinfo=UTC) + completed_at = datetime(2026, 9, 18, 10, 0, 2, tzinfo=UTC) + expected_run_id = "20260918T100001123456Z" + config = object() + client = object() + service = object() + observed: dict[str, object] = {} + settings = SimpleNamespace( + github_token="test-token", + github_api_base_url="https://example.invalid", + github_api_version="test-version", + github_timeout_seconds=1.0, + github_max_retries=0, + github_backoff_seconds=0.0, + log_level="INFO", + ) + + monkeypatch.setattr(pipeline, "load_repository_config", lambda path: config) + monkeypatch.setattr(pipeline.Settings, "from_env", lambda: settings) + monkeypatch.setattr( + pipeline, + "configure_logging", + lambda level: observed.update(log_level=level), + ) + + def build_client(**kwargs: object) -> object: + observed["client_kwargs"] = kwargs + return client + + def build_service(received_client: object) -> object: + observed["service_client"] = received_client + return service + + class FakeBatch: + def __init__(self, received_service: object, *, clock) -> None: + observed["batch_service"] = received_service + observed["batch_times"] = [clock(), clock(), clock()] + + def run(self, received_config: object, *, output_root: Path): + observed["batch_config"] = received_config + observed["output_root"] = output_root + return RepositoryBatchResult( + run_id=expected_run_id, + status="success", + repositories_succeeded=2, + repositories_failed=0, + output_root=output_root, + manifest_path=output_root / "manifest.json", + ) + + monkeypatch.setattr(pipeline, "GitHubClient", build_client) + monkeypatch.setattr(pipeline, "GitHubRepositoryService", build_service) + monkeypatch.setattr(pipeline, "GitHubRepositoryBatch", FakeBatch) + + result = pipeline.run_pipeline( + config_path=tmp_path / "repositories.json", + output_root=tmp_path / "data", + clock=sequence_clock([started_at, extracted_at, completed_at]), + ) + + assert result.status == pipeline.PipelineStatus.SUCCESS + assert result.exit_code == 0 + assert result.run_id == expected_run_id + assert result.extracted_at == extracted_at + assert result.repositories_succeeded == 2 + assert observed["service_client"] is client + assert observed["batch_service"] is service + assert observed["batch_config"] is config + assert observed["batch_times"] == [started_at, extracted_at, completed_at] + assert [stage.to_record() for stage in result.stages] == [ + {"name": "configuration", "status": "success"}, + {"name": "client", "status": "success"}, + {"name": "service", "status": "success"}, + {"name": "batch", "status": "success"}, + ] + + +def test_pipeline_configuration_failure_skips_dependent_stages( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """Invalid repository configuration must fail before credentials are loaded.""" + times = [ + datetime(2026, 9, 18, 11, 0, 0, tzinfo=UTC), + datetime(2026, 9, 18, 11, 0, 1, tzinfo=UTC), + ] + settings_loaded = False + + def fail_config(path: Path) -> object: + raise RepositoryBatchConfigurationError("invalid repository configuration") + + def load_settings() -> object: + nonlocal settings_loaded + settings_loaded = True + raise AssertionError("settings must not be loaded") + + monkeypatch.setattr(pipeline, "load_repository_config", fail_config) + monkeypatch.setattr(pipeline.Settings, "from_env", load_settings) + + result = pipeline.run_pipeline( + config_path=tmp_path / "invalid.json", + output_root=tmp_path / "data", + clock=sequence_clock(times), + ) + + assert settings_loaded is False + assert result.status == pipeline.PipelineStatus.FAILED + assert result.exit_code == 1 + assert result.error_type == "RepositoryBatchConfigurationError" + assert result.error_message == "invalid repository configuration" + assert [stage.status for stage in result.stages] == [ + pipeline.PipelineStageStatus.FAILED, + pipeline.PipelineStageStatus.SKIPPED, + pipeline.PipelineStageStatus.SKIPPED, + pipeline.PipelineStageStatus.SKIPPED, + ] + + +@pytest.mark.parametrize( + ("batch_status", "expected_status", "expected_exit_code"), + [ + ("partial_success", pipeline.PipelineStatus.PARTIAL_SUCCESS, 2), + ("failed", pipeline.PipelineStatus.FAILED, 1), + ], +) +def test_pipeline_preserves_batch_terminal_status( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + batch_status: str, + expected_status: pipeline.PipelineStatus, + expected_exit_code: int, +) -> None: + """Partial and failed batches must remain distinguishable to automation.""" + started_at = datetime(2026, 9, 18, 12, 0, 0, tzinfo=UTC) + extracted_at = datetime(2026, 9, 18, 12, 0, 1, tzinfo=UTC) + completed_at = datetime(2026, 9, 18, 12, 0, 2, tzinfo=UTC) + settings = SimpleNamespace( + github_token="test-token", + github_api_base_url="https://example.invalid", + github_api_version="test-version", + github_timeout_seconds=1.0, + github_max_retries=0, + github_backoff_seconds=0.0, + log_level="INFO", + ) + + monkeypatch.setattr(pipeline, "load_repository_config", lambda path: object()) + monkeypatch.setattr(pipeline.Settings, "from_env", lambda: settings) + monkeypatch.setattr(pipeline, "configure_logging", lambda level: None) + monkeypatch.setattr(pipeline, "GitHubClient", lambda **kwargs: object()) + monkeypatch.setattr(pipeline, "GitHubRepositoryService", lambda client: object()) + + class FakeBatch: + def __init__(self, service: object, *, clock) -> None: + self.clock = clock + + def run(self, config: object, *, output_root: Path): + self.clock() + self.clock() + self.clock() + return RepositoryBatchResult( + run_id="20260918T120001000000Z", + status=batch_status, + repositories_succeeded=int(batch_status == "partial_success"), + repositories_failed=1, + output_root=output_root, + manifest_path=output_root / "manifest.json", + ) + + monkeypatch.setattr(pipeline, "GitHubRepositoryBatch", FakeBatch) + + result = pipeline.run_pipeline( + config_path=tmp_path / "repositories.json", + output_root=tmp_path / "data", + clock=sequence_clock([started_at, extracted_at, completed_at]), + ) + + assert result.status == expected_status + assert result.exit_code == expected_exit_code + + +def test_pipeline_output_failure_does_not_expose_exception_message( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """Unexpected filesystem messages must not leak into automation output.""" + secret = "Authorization: Bearer simulated-token" + times = [ + datetime(2026, 9, 18, 13, 0, 0, tzinfo=UTC), + datetime(2026, 9, 18, 13, 0, 1, tzinfo=UTC), + ] + settings = SimpleNamespace( + github_token="test-token", + github_api_base_url="https://example.invalid", + github_api_version="test-version", + github_timeout_seconds=1.0, + github_max_retries=0, + github_backoff_seconds=0.0, + log_level="INFO", + ) + monkeypatch.setattr(pipeline, "load_repository_config", lambda path: object()) + monkeypatch.setattr(pipeline.Settings, "from_env", lambda: settings) + monkeypatch.setattr(pipeline, "configure_logging", lambda level: None) + monkeypatch.setattr(pipeline, "GitHubClient", lambda **kwargs: object()) + monkeypatch.setattr(pipeline, "GitHubRepositoryService", lambda client: object()) + + class FailingBatch: + def __init__(self, service: object, *, clock) -> None: + pass + + def run(self, config: object, *, output_root: Path): + raise OSError(secret) + + monkeypatch.setattr(pipeline, "GitHubRepositoryBatch", FailingBatch) + + result = pipeline.run_pipeline( + config_path=tmp_path / "repositories.json", + output_root=tmp_path / "data", + clock=sequence_clock(times), + ) + + serialized = result.to_record() + assert serialized["error_message"] == "Pipeline outputs could not be published." + assert secret not in str(serialized) diff --git a/tests/test_repository_batch.py b/tests/test_repository_batch.py index 17cc3fb..648aad2 100644 --- a/tests/test_repository_batch.py +++ b/tests/test_repository_batch.py @@ -227,6 +227,8 @@ def test_batch_persists_raw_normalized_and_manifest_with_shared_timestamp( assert manifest["repositories_requested"] == 2 assert manifest["repositories_succeeded"] == 2 assert manifest["repositories_failed"] == 0 + assert manifest["raw_records_written"] == 2 + assert manifest["normalized_records_written"] == 2 assert manifest["raw_output"].startswith("raw/") assert manifest["normalized_output"].startswith("normalized/") assert "\\" not in manifest["raw_output"] @@ -255,6 +257,7 @@ def test_batch_persists_raw_normalized_and_manifest_with_shared_timestamp( def test_batch_continues_after_failure_and_reports_partial_success( tmp_path: Path, + caplog: pytest.LogCaptureFixture, ) -> None: """One repository failure must not stop subsequent repositories.""" times = [datetime(2026, 7, 23, 13, 0, second, tzinfo=UTC) for second in (0, 1, 2)] @@ -291,6 +294,12 @@ def test_batch_continues_after_failure_and_reports_partial_success( } ] assert service.extract_repository_with_payload.call_count == 2 + failure_record = next( + record + for record in caplog.records + if record.getMessage() == "GitHub repository batch item failed." + ) + assert failure_record.run_id == result.run_id def test_batch_continues_without_orphan_raw_after_serialization_failure( diff --git a/tests/test_repository_metadata.py b/tests/test_repository_metadata.py index 97bf4a5..ab3b816 100644 --- a/tests/test_repository_metadata.py +++ b/tests/test_repository_metadata.py @@ -156,6 +156,20 @@ def test_negative_repository_metric_is_rejected() -> None: RepositoryMetadata.from_github_payload(payload) +def test_zero_repository_id_is_rejected() -> None: + """GitHub repository identifiers must satisfy the analytics contract.""" + payload = { + **VALID_PAYLOAD, + "id": 0, + } + + with pytest.raises( + RepositoryMetadataError, + match="id.*greater than zero", + ): + RepositoryMetadata.from_github_payload(payload) + + def test_naive_extraction_datetime_is_rejected() -> None: """The extraction timestamp must include timezone information.""" naive_datetime = datetime( diff --git a/tests/test_run_devflow.py b/tests/test_run_devflow.py new file mode 100644 index 0000000..b2331bb --- /dev/null +++ b/tests/test_run_devflow.py @@ -0,0 +1,104 @@ +"""Tests for the DevFlow pipeline command-line interface.""" + +from __future__ import annotations + +import json +import logging +from datetime import UTC, datetime +from pathlib import Path + +import pytest + +from ingestion.common.logging import configure_logging +from orchestration.pipeline import PipelineResult, PipelineStatus +from scripts import run_devflow + + +@pytest.mark.parametrize( + ("status", "expected_code"), + [ + (PipelineStatus.SUCCESS, 0), + (PipelineStatus.PARTIAL_SUCCESS, 2), + (PipelineStatus.FAILED, 1), + ], +) +def test_main_prints_json_result_and_returns_pipeline_exit_code( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + status: PipelineStatus, + expected_code: int, +) -> None: + """The module entry point must provide stable machine-readable output.""" + started_at = datetime(2026, 9, 18, 14, 0, 0, tzinfo=UTC) + extracted_at = datetime(2026, 9, 18, 14, 0, 1, tzinfo=UTC) + observed: dict[str, Path] = {} + + def fake_run_pipeline(*, config_path: Path, output_root: Path) -> PipelineResult: + observed["config_path"] = config_path + observed["output_root"] = output_root + return PipelineResult( + run_id="20260918T140001000000Z", + started_at=started_at, + extracted_at=extracted_at, + status=status, + repositories_succeeded=int(status != PipelineStatus.FAILED), + repositories_failed=int(status != PipelineStatus.SUCCESS), + manifest_path=output_root / "manifest.json", + stages=(), + ) + + monkeypatch.setattr(run_devflow, "run_pipeline", fake_run_pipeline) + config_path = tmp_path / "repositories.json" + output_root = tmp_path / "data" + + exit_code = run_devflow.main( + ["--config", str(config_path), "--output-root", str(output_root)] + ) + + output = json.loads(capsys.readouterr().out) + assert exit_code == expected_code + assert observed == {"config_path": config_path, "output_root": output_root} + assert output["status"] == status.value + assert output["exit_code"] == expected_code + assert output["run_id"] == "20260918T140001000000Z" + + +def test_main_keeps_structured_logs_out_of_json_stdout( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """Logs must use stderr so stdout remains one machine-readable document.""" + started_at = datetime(2026, 9, 18, 15, 0, 0, tzinfo=UTC) + configure_logging() + + def fake_run_pipeline(*, config_path: Path, output_root: Path) -> PipelineResult: + logging.getLogger("devflow.test").warning("pipeline diagnostic") + return PipelineResult( + run_id="20260918T150000000000Z", + started_at=started_at, + extracted_at=started_at, + status=PipelineStatus.SUCCESS, + repositories_succeeded=1, + repositories_failed=0, + manifest_path=output_root / "manifest.json", + stages=(), + ) + + monkeypatch.setattr(run_devflow, "run_pipeline", fake_run_pipeline) + + exit_code = run_devflow.main( + [ + "--config", + str(tmp_path / "repositories.json"), + "--output-root", + str(tmp_path / "data"), + ] + ) + + captured = capsys.readouterr() + assert exit_code == 0 + assert json.loads(captured.out)["status"] == "success" + assert "pipeline diagnostic" not in captured.out + assert "pipeline diagnostic" in captured.err From da4b350d70c4447a109ced96c69a7d61ba4f8f30 Mon Sep 17 00:00:00 2001 From: Daniel Burbano Date: Fri, 18 Sep 2026 16:24:35 -0500 Subject: [PATCH 7/8] feat: add BigQuery repository sink --- ingestion/bigquery/__init__.py | 1 + ingestion/bigquery/repository_sink.py | 146 +++++++++++++++++++++++++ ingestion/common/repository_sink.py | 28 +++++ tests/test_bigquery_repository_sink.py | 141 ++++++++++++++++++++++++ 4 files changed, 316 insertions(+) create mode 100644 ingestion/bigquery/__init__.py create mode 100644 ingestion/bigquery/repository_sink.py create mode 100644 ingestion/common/repository_sink.py create mode 100644 tests/test_bigquery_repository_sink.py diff --git a/ingestion/bigquery/__init__.py b/ingestion/bigquery/__init__.py new file mode 100644 index 0000000..8832d48 --- /dev/null +++ b/ingestion/bigquery/__init__.py @@ -0,0 +1 @@ +"""BigQuery-ready ingestion adapters.""" diff --git a/ingestion/bigquery/repository_sink.py b/ingestion/bigquery/repository_sink.py new file mode 100644 index 0000000..0b5e266 --- /dev/null +++ b/ingestion/bigquery/repository_sink.py @@ -0,0 +1,146 @@ +"""BigQuery adapter for normalized GitHub repository records. + +The adapter depends on a small client protocol instead of the Google SDK. +Production composition can inject ``google.cloud.bigquery.Client`` without +making the core ingestion package depend on that library. +""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from datetime import UTC, datetime +from typing import Any, Protocol + +from ingestion.common.repository_sink import RepositoryWriteResult + +REPOSITORY_TABLE_SCHEMA: tuple[dict[str, str], ...] = ( + {"name": "run_id", "type": "STRING", "mode": "REQUIRED"}, + {"name": "repository_id", "type": "INTEGER", "mode": "REQUIRED"}, + {"name": "repository_name", "type": "STRING", "mode": "REQUIRED"}, + {"name": "full_name", "type": "STRING", "mode": "REQUIRED"}, + {"name": "owner_login", "type": "STRING", "mode": "REQUIRED"}, + {"name": "description", "type": "STRING", "mode": "NULLABLE"}, + {"name": "visibility", "type": "STRING", "mode": "REQUIRED"}, + {"name": "default_branch", "type": "STRING", "mode": "REQUIRED"}, + {"name": "language", "type": "STRING", "mode": "NULLABLE"}, + {"name": "is_fork", "type": "BOOLEAN", "mode": "REQUIRED"}, + {"name": "is_archived", "type": "BOOLEAN", "mode": "REQUIRED"}, + {"name": "is_disabled", "type": "BOOLEAN", "mode": "REQUIRED"}, + {"name": "created_at", "type": "TIMESTAMP", "mode": "REQUIRED"}, + {"name": "updated_at", "type": "TIMESTAMP", "mode": "REQUIRED"}, + {"name": "pushed_at", "type": "TIMESTAMP", "mode": "NULLABLE"}, + {"name": "stars_count", "type": "INTEGER", "mode": "REQUIRED"}, + {"name": "forks_count", "type": "INTEGER", "mode": "REQUIRED"}, + {"name": "open_issues_count", "type": "INTEGER", "mode": "REQUIRED"}, + {"name": "subscribers_count", "type": "INTEGER", "mode": "REQUIRED"}, + {"name": "size_kb", "type": "INTEGER", "mode": "REQUIRED"}, + {"name": "html_url", "type": "STRING", "mode": "REQUIRED"}, + {"name": "extracted_at", "type": "TIMESTAMP", "mode": "REQUIRED"}, +) + +_SCHEMA_FIELDS = frozenset(field["name"] for field in REPOSITORY_TABLE_SCHEMA) +_TIMESTAMP_FIELDS = frozenset( + field["name"] for field in REPOSITORY_TABLE_SCHEMA if field["type"] == "TIMESTAMP" +) + + +class BigQueryInsertClient(Protocol): + """Subset of the BigQuery client needed by this adapter.""" + + def insert_rows_json( + self, + table: str, + json_rows: Sequence[Mapping[str, Any]], + *, + row_ids: Sequence[str], + ) -> Sequence[Mapping[str, Any]]: + """Insert JSON rows and return per-row errors.""" + ... + + +class BigQueryRepositoryWriteError(RuntimeError): + """Raised when BigQuery rejects one or more repository rows.""" + + +class BigQueryRepositorySink: + """Write normalized repository records through a compatible client.""" + + def __init__(self, client: BigQueryInsertClient, *, table: str) -> None: + normalized_table = table.strip() + if not normalized_table: + raise ValueError("BigQuery table must be a non-empty identifier.") + self._client = client + self._table = normalized_table + + def write( + self, + records: Sequence[Mapping[str, Any]], + *, + run_id: str, + ) -> RepositoryWriteResult: + """Insert rows with deterministic IDs for best-effort deduplication.""" + normalized_run_id = run_id.strip() + if not normalized_run_id: + raise ValueError("run_id must be non-empty.") + + rows = [_prepare_row(record, run_id=normalized_run_id) for record in records] + row_ids = [f"{normalized_run_id}:{row['repository_id']}" for row in rows] + if len(row_ids) != len(set(row_ids)): + raise ValueError("Repository records contain duplicate identities.") + if not rows: + return RepositoryWriteResult(rows_attempted=0, rows_written=0) + + errors = self._client.insert_rows_json( + self._table, + rows, + row_ids=row_ids, + ) + if errors: + raise BigQueryRepositoryWriteError( + f"BigQuery rejected {len(errors)} repository row(s)." + ) + + return RepositoryWriteResult( + rows_attempted=len(rows), + rows_written=len(rows), + ) + + +def _prepare_row(record: Mapping[str, Any], *, run_id: str) -> dict[str, Any]: + """Validate the storage shape and serialize timestamps as UTC RFC 3339.""" + supported_input_fields = _SCHEMA_FIELDS - {"run_id"} + unknown_fields = set(record) - supported_input_fields + if unknown_fields: + raise ValueError( + "Repository record contains unsupported fields: " + + ", ".join(sorted(unknown_fields)) + ) + + required_fields = { + field["name"] + for field in REPOSITORY_TABLE_SCHEMA + if field["mode"] == "REQUIRED" and field["name"] != "run_id" + } + missing_fields = required_fields - set(record) + if missing_fields: + raise ValueError( + "Repository record is missing required fields: " + + ", ".join(sorted(missing_fields)) + ) + + row = {"run_id": run_id} + for field_name in supported_input_fields: + value = record.get(field_name) + if field_name in _TIMESTAMP_FIELDS and value is not None: + value = _serialize_timestamp(value, field_name=field_name) + row[field_name] = value + return row + + +def _serialize_timestamp(value: object, *, field_name: str) -> str: + """Serialize an aware datetime to a stable UTC BigQuery timestamp value.""" + if not isinstance(value, datetime): + raise ValueError(f"Field '{field_name}' must be a datetime.") + if value.tzinfo is None or value.utcoffset() is None: + raise ValueError(f"Field '{field_name}' must include timezone information.") + return value.astimezone(UTC).isoformat().replace("+00:00", "Z") diff --git a/ingestion/common/repository_sink.py b/ingestion/common/repository_sink.py new file mode 100644 index 0000000..6b615a3 --- /dev/null +++ b/ingestion/common/repository_sink.py @@ -0,0 +1,28 @@ +"""Storage contract for normalized repository records.""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from typing import Any, Protocol + + +@dataclass(frozen=True, slots=True) +class RepositoryWriteResult: + """Summary returned by a repository record sink.""" + + rows_attempted: int + rows_written: int + + +class RepositoryRecordSink(Protocol): + """Destination-independent contract for repository records.""" + + def write( + self, + records: Sequence[Mapping[str, Any]], + *, + run_id: str, + ) -> RepositoryWriteResult: + """Persist normalized records for one extraction run.""" + ... diff --git a/tests/test_bigquery_repository_sink.py b/tests/test_bigquery_repository_sink.py new file mode 100644 index 0000000..7217293 --- /dev/null +++ b/tests/test_bigquery_repository_sink.py @@ -0,0 +1,141 @@ +"""Tests for the dependency-free BigQuery repository sink.""" + +from datetime import UTC, datetime, timedelta, timezone +from unittest.mock import Mock + +import pytest + +from ingestion.bigquery.repository_sink import ( + REPOSITORY_TABLE_SCHEMA, + BigQueryInsertClient, + BigQueryRepositorySink, + BigQueryRepositoryWriteError, +) + + +def build_record() -> dict[str, object]: + """Build one normalized repository record.""" + return { + "repository_id": 1296269, + "repository_name": "airflow", + "full_name": "apache/airflow", + "owner_login": "apache", + "description": None, + "visibility": "public", + "default_branch": "main", + "language": "Python", + "is_fork": False, + "is_archived": False, + "is_disabled": False, + "created_at": datetime( + 2015, + 4, + 13, + 13, + 4, + 58, + tzinfo=timezone(timedelta(hours=-5)), + ), + "updated_at": datetime(2026, 7, 22, 20, 0, tzinfo=UTC), + "pushed_at": None, + "stars_count": 40000, + "forks_count": 15000, + "open_issues_count": 1000, + "subscribers_count": 700, + "size_kb": 250000, + "html_url": "https://github.com/apache/airflow", + "extracted_at": datetime(2026, 7, 22, 22, 0, tzinfo=UTC), + } + + +def test_schema_matches_normalized_repository_contract() -> None: + """The public schema must remain explicit and complete.""" + fields = {field["name"]: field for field in REPOSITORY_TABLE_SCHEMA} + + assert set(fields) == {"run_id", *build_record()} + assert fields["repository_id"] == { + "name": "repository_id", + "type": "INTEGER", + "mode": "REQUIRED", + } + assert fields["pushed_at"]["mode"] == "NULLABLE" + assert fields["extracted_at"]["type"] == "TIMESTAMP" + + +def test_write_uses_utc_timestamps_and_deterministic_row_ids() -> None: + """Rows must be BigQuery-ready and retry with the same insertion identity.""" + client = Mock(spec=BigQueryInsertClient) + client.insert_rows_json.return_value = [] + sink = BigQueryRepositorySink(client, table="project.dataset.repositories") + + result = sink.write([build_record()], run_id="run-123") + + assert result.rows_attempted == 1 + assert result.rows_written == 1 + table, rows = client.insert_rows_json.call_args.args + assert table == "project.dataset.repositories" + assert rows[0]["run_id"] == "run-123" + assert rows[0]["created_at"] == "2015-04-13T18:04:58Z" + assert rows[0]["extracted_at"] == "2026-07-22T22:00:00Z" + assert client.insert_rows_json.call_args.kwargs["row_ids"] == ["run-123:1296269"] + + +def test_write_rejects_naive_datetime_before_client_call() -> None: + """Naive timestamps must never reach BigQuery with ambiguous semantics.""" + client = Mock(spec=BigQueryInsertClient) + record = build_record() + record["updated_at"] = datetime(2026, 7, 22, 20, 0) + + with pytest.raises(ValueError, match="updated_at.*timezone"): + BigQueryRepositorySink(client, table="project.dataset.repositories").write( + [record], + run_id="run-123", + ) + + client.insert_rows_json.assert_not_called() + + +def test_write_rejects_duplicate_repository_identity() -> None: + """One run cannot submit duplicate insertion IDs.""" + client = Mock(spec=BigQueryInsertClient) + record = build_record() + + with pytest.raises(ValueError, match="duplicate identities"): + BigQueryRepositorySink(client, table="project.dataset.repositories").write( + [record, record.copy()], + run_id="run-123", + ) + + client.insert_rows_json.assert_not_called() + + +def test_write_raises_sanitized_error_when_bigquery_rejects_rows() -> None: + """Provider error bodies must not be copied into application exceptions.""" + client = Mock(spec=BigQueryInsertClient) + client.insert_rows_json.return_value = [ + {"index": 0, "errors": [{"message": "Authorization: simulated-secret"}]} + ] + + with pytest.raises(BigQueryRepositoryWriteError) as exception_info: + BigQueryRepositorySink(client, table="project.dataset.repositories").write( + [build_record()], + run_id="run-123", + ) + + message = str(exception_info.value) + assert "simulated-secret" not in message + assert "Authorization" not in message + + +def test_empty_batch_does_not_call_client() -> None: + """An empty batch is a successful no-op.""" + client = Mock(spec=BigQueryInsertClient) + + result = BigQueryRepositorySink( + client, + table="project.dataset.repositories", + ).write([], run_id="run-123") + + assert result.rows_attempted == 0 + assert result.rows_written == 0 + client.insert_rows_json.assert_not_called() From 559879977393edeb73348e4e27d78a09da48c460 Mon Sep 17 00:00:00 2001 From: Daniel Burbano Date: Fri, 18 Sep 2026 16:24:54 -0500 Subject: [PATCH 8/8] chore: add CI and secret-safe logging --- .github/workflows/ci.yml | 39 +++++++++++++++++++++++ .gitignore | 3 +- ingestion/common/logging.py | 58 +++++++++++++++++++++++++++++++--- tests/test_logging.py | 63 +++++++++++++++++++++++++++++++++++++ 4 files changed, 157 insertions(+), 6 deletions(-) create mode 100644 .github/workflows/ci.yml diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..e9d4490 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,39 @@ +name: CI + +on: + push: + pull_request: + +permissions: + contents: read + +jobs: + quality: + runs-on: windows-latest + timeout-minutes: 15 + + steps: + - name: Check out repository + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + cache: pip + cache-dependency-path: requirements-lock.txt + + - name: Install locked dependencies + run: python -m pip install --disable-pip-version-check -r requirements-lock.txt + + - name: Compile Python sources + run: python -m compileall ingestion orchestration scripts + + - name: Ruff lint + run: python -m ruff check . + + - name: Ruff format check + run: python -m ruff format --check . + + - name: Test + run: python -m pytest diff --git a/.gitignore b/.gitignore index 6048d3a..526218a 100644 --- a/.gitignore +++ b/.gitignore @@ -11,7 +11,8 @@ venv/ env/ # Variables de entorno y credenciales -.env +.env* +!.env.example *.key *.pem credentials.json diff --git a/ingestion/common/logging.py b/ingestion/common/logging.py index ef80a9f..a0d62af 100644 --- a/ingestion/common/logging.py +++ b/ingestion/common/logging.py @@ -4,10 +4,34 @@ import json import logging +import re import sys +from collections.abc import Mapping from datetime import UTC, datetime from typing import Any +_REDACTED = "[REDACTED]" +_SENSITIVE_KEY_PARTS = ( + "api-key", + "api_key", + "apikey", + "authorization", + "credential", + "password", + "secret", + "token", +) +_SENSITIVE_TEXT_PATTERNS = ( + re.compile( + r"(?i)\b(authorization[\"']?\s*[:=]\s*[\"']?)" + r"(?:bearer\s+)?[^\s,;}\"']+" + ), + re.compile( + r"(?i)\b((?:api[-_\s]?key|credential|password|secret|token)" + r"[\"']?\s*[:=]\s*[\"']?)[^\s,;}\"']+" + ), +) + _STANDARD_LOG_RECORD_FIELDS = frozenset(logging.makeLogRecord({}).__dict__) | frozenset( { "message", @@ -28,7 +52,7 @@ def format(self, record: logging.LogRecord) -> str: ).isoformat(), "level": record.levelname, "logger": record.name, - "message": record.getMessage(), + "message": _redact_text(record.getMessage()), } for key, value in record.__dict__.items(): @@ -38,10 +62,10 @@ def format(self, record: logging.LogRecord) -> str: if key.startswith("_"): continue - payload[key] = value + payload[key] = _redact(value, key=key) if record.exc_info: - payload["exception"] = self.formatException(record.exc_info) + payload["exception"] = _redact_text(self.formatException(record.exc_info)) return json.dumps( payload, @@ -50,15 +74,39 @@ def format(self, record: logging.LogRecord) -> str: ) +def _redact(value: Any, *, key: str | None = None) -> Any: + """Recursively redact values whose field names commonly contain secrets.""" + if key is not None and any(part in key.casefold() for part in _SENSITIVE_KEY_PARTS): + return _REDACTED + if isinstance(value, Mapping): + return { + item_key: _redact(item, key=str(item_key)) + for item_key, item in value.items() + } + if isinstance(value, (list, tuple)): + return [_redact(item) for item in value] + if isinstance(value, str): + return _redact_text(value) + return value + + +def _redact_text(value: str) -> str: + """Redact common inline secret assignments from free-form text.""" + redacted = value + for pattern in _SENSITIVE_TEXT_PATTERNS: + redacted = pattern.sub(r"\1[REDACTED]", redacted) + return redacted + + def configure_logging(level: str = "INFO") -> None: - """Configure application logging to write JSON records to stdout.""" + """Configure application logging to write JSON records to stderr.""" normalized_level = level.strip().upper() numeric_level = getattr(logging, normalized_level, None) if not isinstance(numeric_level, int): raise ValueError(f"Invalid logging level: {level}") - handler = logging.StreamHandler(sys.stdout) + handler = logging.StreamHandler(sys.stderr) handler.setFormatter(JsonFormatter()) root_logger = logging.getLogger() diff --git a/tests/test_logging.py b/tests/test_logging.py index 7df4cde..3853856 100644 --- a/tests/test_logging.py +++ b/tests/test_logging.py @@ -2,6 +2,7 @@ import json import logging +import sys from ingestion.common.logging import JsonFormatter @@ -30,3 +31,65 @@ def test_json_formatter_includes_context_fields() -> None: assert payload["operation"] == "github_get" assert payload["status_code"] == 200 assert "timestamp" in payload + + +def test_json_formatter_redacts_nested_secret_fields() -> None: + """Structured context must redact common credential field names.""" + record = logging.LogRecord( + name="ingestion.bigquery", + level=logging.ERROR, + pathname=__file__, + lineno=20, + msg="Write failed.", + args=(), + exc_info=None, + ) + record.run_id = "run-123" + record.context = { + "access_token": "simulated-token", + "api_key": "simulated-api-key", + "nested": [{"Authorization": "Bearer simulated-secret"}], + "safe": "visible", + } + + payload = json.loads(JsonFormatter().format(record)) + + assert payload["run_id"] == "run-123" + assert payload["context"]["access_token"] == "[REDACTED]" + assert payload["context"]["api_key"] == "[REDACTED]" + assert payload["context"]["nested"][0]["Authorization"] == "[REDACTED]" + assert payload["context"]["safe"] == "visible" + assert "simulated-token" not in json.dumps(payload) + assert "simulated-api-key" not in json.dumps(payload) + assert "simulated-secret" not in json.dumps(payload) + + +def test_json_formatter_redacts_inline_secrets_from_message_and_exception() -> None: + """Free-form messages and tracebacks must not expose assigned secrets.""" + try: + raise RuntimeError("Authorization: Bearer exception-secret") + except RuntimeError: + record = logging.LogRecord( + name="ingestion.bigquery", + level=logging.ERROR, + pathname=__file__, + lineno=50, + msg=( + "Request failed with token=message-secret api-key: secondary-secret " + 'payload={"api_key":"json-secret"}' + ), + args=(), + exc_info=sys.exc_info(), + ) + + payload = json.loads(JsonFormatter().format(record)) + + assert payload["message"] == ( + "Request failed with token=[REDACTED] api-key: [REDACTED] " + 'payload={"api_key":"[REDACTED]"}' + ) + assert "Authorization: [REDACTED]" in payload["exception"] + assert "message-secret" not in json.dumps(payload) + assert "secondary-secret" not in json.dumps(payload) + assert "json-secret" not in json.dumps(payload) + assert "exception-secret" not in json.dumps(payload)