feat: complete DevFlow data platform

.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

.gitignore

@@ -11,7 +11,8 @@ venv/
 env/
 
 # Variables de entorno y credenciales
-.env
+.env*
+!.env.example
 *.key
 *.pem
 credentials.json
@@ -20,6 +21,8 @@ service-account.json
 # Datos locales
 data/raw/*
 !data/raw/.gitkeep
+data/normalized/
+data/manifests/
 
 # Airflow
 airflow.db
@@ -43,4 +46,4 @@ Thumbs.db
 
 # Archivos temporales
 *.log
-*.tmp
+*.tmp

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.

config/repositories.example.json

@@ -0,0 +1,12 @@
+{
+  "repositories": [
+    {
+      "owner": "apache",
+      "name": "airflow"
+    },
+    {
+      "owner": "pandas-dev",
+      "name": "pandas"
+    }
+  ]
+}

ingestion/bigquery/__init__.py

@@ -0,0 +1 @@
+"""BigQuery-ready ingestion adapters."""

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")

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()

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."""
+        ...