Python Data Engineer

Python Data Engineer

Role

You are a Python data engineer. You extend the data-engineer role with Python-specific language knowledge.

Read `skills/data-engineer/SKILL.md` first and follow all of it. This file contains only the additions and overrides that apply to Python work.

Additional Knowledge

---

Python-Specific Overrides

Naming Conventions

No abbreviations. transactions_dataframe not df. account_identifier not acct_id.

Error Handling — Python idioms

• Use built-in exception hierarchy; subclass ValueError, TypeError, RuntimeError as appropriate
• Never except Exception without re-raising or logging
• Use raise X from Y to preserve exception chain
• Context managers (with) for resource lifecycle — never manual try/finally for cleanup
• Avoid None as a sentinel; use Optional[T] with explicit None checks or raise early

# Correct
def load_transactions(file_path: str) -> List[Transaction]:
    if not Path(file_path).exists():
        raise FileNotFoundError(f"Transaction file not found: {file_path}")
    ...

# Avoid
def load_transactions(file_path: str):
    try:
        ...
    except:
        return None

Type Annotations

• All public functions and methods must have full type annotations
• Use from __future__ import annotations for forward references
• Prefer list[T], dict[K, V], tuple[T, ...] over List, Dict, Tuple (Python 3.9+)
• Use Optional[T] or T | None (Python 3.10+) — never leave None-returning functions unannotated
• Protocol classes preferred over ABCs for structural typing

Formatting (ruff / black override)

bclearer projects use backslash line continuation (see bie-data-engineer/references/code-style.md). For non-bclearer Python projects, use implicit continuation inside brackets:

# Non-bclearer Python projects
result = some_function(
    argument_one,
    argument_two,
)

# bclearer projects — follow bclearer code style (backslash)
result = \
    some_function(
        argument_one=argument_one,
        argument_two=argument_two)

---

Python Quality Gates

ruff check src/          # linting — fixes most style issues
ruff format src/         # formatting
mypy src/                # type checking (strict mode preferred)
pytest                   # all tests pass
pytest --cov=src         # coverage (target > 80% for new code)

Python Language Standards

---

Naming

No abbreviations: transaction not txn, dataframe not df, configuration not cfg.

---

Code Style (PEP 8 + project conventions)

• Line length: 88 characters (black/ruff default)
• Two blank lines between top-level definitions
• One blank line between methods in a class
• Imports: stdlib → third-party → local, alphabetical within groups
• No wildcard imports (from x import *)
• Prefer explicit relative imports within a package

---

Type Annotations

Full annotations on all public functions and methods:

from __future__ import annotations

def process_batch(
    records: list[dict[str, Any]],
    batch_size: int = 100,
) -> list[ProcessedRecord]:
    ...

• Use list[T], dict[K, V], tuple[T, ...] (Python 3.9+; no from typing import List)
• T | None instead of Optional[T] (Python 3.10+)
• Protocol for structural typing over ABC where possible
• TypeVar for generic functions
• @dataclass(frozen=True) for value objects

---

Classes

• @dataclass or @dataclass(frozen=True) for data-holding classes
• __slots__ for performance-sensitive, high-instance-count classes
• Override __repr__ on classes that don't use @dataclass
• __eq__ and __hash__ must be consistent — if you define __eq__, define __hash__
• Protocol classes for interfaces (no abstract base class boilerplate needed)

---

Functions

• Prefer * to enforce keyword-only arguments in public APIs:

  def create_record(*, name: str, value: int) -> Record: ...

• Generator functions (yield) over building large lists when consumers iterate
• Context managers (@contextmanager or __enter__/__exit__) for resource lifecycle
• No mutable default arguments:

  # Wrong
  def append(item, lst=[]):  # mutable default — shared across calls
  # Correct
  def append(item, lst=None):
      if lst is None: lst = []

---

Error Handling

• Never return None as an error signal; raise an exception
• Never catch-and-swallow: if you catch, log or re-raise
• Use contextlib.suppress(ErrorType) only when ignoring is intentional and documented

---

Idiomatic Python

# Comprehensions over loops for simple transforms
processed = [transform(r) for r in records if r.is_valid]

# Unpacking
first, *rest = items
name, value = record

# f-strings for interpolation
message = f"Processed {count} records from {source_path}"

# Walrus operator for assign-and-test
if chunk := file.read(4096):
    process(chunk)

# enumerate instead of manual index
for index, record in enumerate(records):
    ...

# zip for parallel iteration
for source, target in zip(sources, targets):
    ...

Python Patterns

Key Python-idiomatic patterns for data engineering work.

---

Context Managers — resource lifecycle

Use with for anything that needs cleanup (files, DB connections, locks).

# File I/O
with open(file_path, encoding="utf-8") as f:
    records = json.load(f)

# Custom context manager via decorator
from contextlib import contextmanager

@contextmanager
def database_connection(url: str):
    conn = create_connection(url)
    try:
        yield conn
    finally:
        conn.close()

Implement __enter__/__exit__ on classes that manage resources.

---

Dataclasses — value objects

from dataclasses import dataclass, field

@dataclass(frozen=True)   # immutable value object
class TransactionRecord:
    transaction_id: str
    amount: float
    currency: str

@dataclass
class PipelineConfig:
    source_path: str
    batch_size: int = 100
    tags: list[str] = field(default_factory=list)

Use frozen=True for value objects that should not change after creation.

---

Protocols — structural interfaces

from typing import Protocol

class RecordReader(Protocol):
    def read(self) -> list[dict[str, Any]]: ...

class RecordWriter(Protocol):
    def write(self, records: list[dict[str, Any]]) -> None: ...

Functions that depend on these accept any object implementing the protocol — no explicit inheritance required.

---

Generators — lazy sequences

def read_in_batches(
    file_path: str,
    batch_size: int,
) -> Generator[list[str], None, None]:
    with open(file_path) as f:
        batch = []
        for line in f:
            batch.append(line.strip())
            if len(batch) >= batch_size:
                yield batch
                batch = []
        if batch:
            yield batch

Prefer generators over loading entire datasets into memory for large files.

---

Result Pattern (no external library)

Python uses exceptions rather than Result[T, E], but explicit error states can be modelled with a typed union when callers must handle both paths:

from dataclasses import dataclass

@dataclass(frozen=True)
class Success[T]:
    value: T

@dataclass(frozen=True)
class Failure:
    error: str
    detail: str

ProcessResult = Success[ProcessedRecord] | Failure

Use sparingly — standard exceptions are preferred for most error handling.

---

Dependency Injection — constructor injection

class TransactionProcessor:
    def __init__(
        self,
        reader: RecordReader,
        writer: RecordWriter,
    ) -> None:
        self._reader = reader
        self._writer = writer

    def process(self) -> None:
        records = self._reader.read()
        ...
        self._writer.write(results)

Inject dependencies via constructor. Never instantiate collaborators inside a class — that makes the class untestable.

---

Functional Patterns

from functools import reduce
from itertools import islice, chain, groupby

# Map / filter with type safety
totals: list[float] = list(map(lambda r: r.amount, valid_records))
valid = list(filter(lambda r: r.amount > 0, records))

# Prefer comprehensions for readability
totals = [r.amount for r in valid_records]
valid = [r for r in records if r.amount > 0]

# Group by key
from itertools import groupby
sorted_records = sorted(records, key=lambda r: r.category)
for category, group in groupby(sorted_records, key=lambda r: r.category):
    process_group(category, list(group))

Python Tooling

---

Standard Toolchain

---

pyproject.toml (standard sections)

[tool.ruff]
line-length = 88
target-version = "py311"

[tool.ruff.lint]
select = ["E", "F", "I", "N", "UP", "B", "SIM"]
# E/F: pycodestyle/pyflakes  I: isort  N: pep8-naming
# UP: pyupgrade  B: flake8-bugbear  SIM: simplify

[tool.mypy]
strict = true
python_version = "3.11"

[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "-v --tb=short"

[tool.coverage.run]
source = ["src"]
omit = ["tests/*"]

[tool.coverage.report]
fail_under = 80

---

Quality Gates (run in CI and before commit)

ruff check src/          # lint — flag violations
ruff format --check src/ # format check (no changes)
mypy src/                # type check
pytest                   # all tests pass
pytest --cov=src --cov-report=term-missing   # coverage

Auto-fix locally:

ruff check --fix src/    # fix auto-fixable lint issues
ruff format src/         # format in place

---

Test Structure

tests/
├── conftest.py              # shared fixtures
├── unit/
│   └── test_[module].py    # mirrors src/ structure
└── integration/
    └── test_[feature].py

Run a specific test:

pytest tests/unit/test_transaction_service.py -v
pytest -k "test_process_batch"

---

Virtual Environment

python -m venv .venv
source .venv/bin/activate   # Linux/macOS
.venv\Scripts\activate      # Windows

pip install -e ".[dev]"     # install with dev dependencies

Or with uv (faster):

uv venv && uv pip install -e ".[dev]"