Menü
Team conventions for Python data pipelines — stage structure, JSON output format, debugging workflow, and anti-patterns. Supplements standard patterns with team-specific rules.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Basierend auf der SOC-Berufsklassifikation
Use when writing Python error handling code. Provides specific rules for exception chaining and custom error hierarchies.
Use when reviewing code for security issues.
Use when implementing features, fixing bugs, or refactoring code that may invalidate existing docs. Detects stale documentation by matching the code diff against in-repo doc files and applies targeted updates. Relevant when code changes rename, remove, or add APIs, fields, config keys, or CLI flags. Also applies when the user says "update docs", "check docs", or "are the docs stale", or when reviewing a branch for documentation accuracy.
Helps you write better Python code by following clean code principles and software engineering best practices.
Unified verification engine for Python data science projects. Covers environment checks, type checking, linting, tests, security scans, code review with DS anti-patterns, and notebook checks. Commands (/verify, /quality-gate) invoke different subsets of this skill.
Use when the user asks to design, plan, or explore approaches before implementing — creating features, building components, or adding functionality that would benefit from design exploration first.
| name | data-pipeline-patterns |
| version | 1.0 |
| description | Team conventions for Python data pipelines — stage structure, JSON output format, debugging workflow, and anti-patterns. Supplements standard patterns with team-specific rules. |
Claude already knows validation, error handling, and retry patterns. This covers our team's specific stage structure and conventions.
Every pipeline stage follows this pattern:
def main(argv=None):
args = parse_args(argv)
# 1. Load input
input_data = load_json_file(Path(args.file))
if input_data is None:
sys.exit(1)
# 2. Validate input
if not validate_json_structure(input_data, ["required_key"], "Input file"):
sys.exit(1)
# 3. Process
result = process(input_data)
# 4. Save output with metadata
output = {
"metadata": {
"source_file": str(args.file),
"generated_at": datetime.now().isoformat(),
"items_processed": len(result),
},
"data": result
}
save_json_file(output, output_path)
Key rules: Each stage is independently runnable (CLI + importable). Validate input before processing. Include metadata in every output. Fail fast on invalid input. Checkpoint every N items on long operations.
metadata dict + data (list or dict)source_file, generated_at, count fields_ prefix for derived/computed fields2026-03-15T10:00:00)indent=2 and ensure_ascii=False| Symptom | Likely Cause | Fix |
|---|---|---|
| Empty output file | Input had no matching items | Check filters, validate input data |
| Missing keys in output | Schema changed upstream | Update validation, check input stage |
| API timeout | Service overloaded or VPN down | Add retry logic, check connectivity |
| Rate limit errors (429) | Too many API calls | Add backoff, reduce batch size |
| Partial output | Stage crashed mid-processing | Add checkpointing |
| Wrong data types | API returned unexpected format | Add type validation at boundaries |
| Duplicate items | Pagination overlap | Deduplicate by key field |
except: pass in pipeline stages