api-integration-patterns

API integration patterns for autonomous-dev including subprocess safety, GitHub CLI integration, retry logic, authentication, rate limiting, and timeout handling. Use when integrating external APIs or CLI tools.

Zur Installation springen

Repository: diegosouzapw/awesome-omni-skill
Letzte Quellaktivität: 1. März 2026 um 06:20
Erkannte Sprache von SKILL.md: Englisch
Sterne: 50
Forks: 19

Installationsoptionen

Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.

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.

Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.

npx skills add https://github.com/diegosouzapw/awesome-omni-skill --skill api-integration-patterns

Der Befehl bleibt in einer Zeile. Scrollen Sie horizontal, um ihn vor dem Kopieren vollständig zu prüfen.

Sie bevorzugen eine lokale Kopie? Laden Sie die Dateien herunter, die SkillsMP derzeit vorliegen.

Quelldateien prüfen

Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.

Datei-Explorer

2 Dateien

SKILL.md

Quellanweisungen · Schreibgeschützte Vorschau

Zurück zur Suche Zurück zur Installation

import subprocess
from typing import List

def safe_subprocess(
    command: List[str],
    *,
    allowed_commands: List[str],
    timeout: int = 30
) -> subprocess.CompletedProcess:
    """Execute subprocess with CWE-78 prevention.

    Args:
        command: Command and arguments as list (NOT string!)
        allowed_commands: Whitelist of allowed commands
        timeout: Maximum execution time in seconds

    Returns:
        Completed subprocess result

    Raises:
        SecurityError: If command not in whitelist
        subprocess.TimeoutExpired: If timeout exceeded

    Security:
        - CWE-78 Prevention: Argument arrays (no shell injection)
        - Command Whitelist: Only approved commands
        - Timeout: DoS prevention

    Example:
        >>> result = safe_subprocess(
        ...     ["gh", "issue", "create", "--title", user_title],
        ...     allowed_commands=["gh", "git"]
        ... )
    """
    # Whitelist validation
    if command[0] not in allowed_commands:
        raise SecurityError(f"Command not allowed: {command[0]}")

    # Execute with argument array (NEVER shell=True!)
    return subprocess.run(
        command,
        capture_output=True,
        text=True,
        timeout=timeout,
        check=True,
        shell=False  # CRITICAL
    )

def create_github_issue(
    title: str,
    body: str,
    *,
    labels: Optional[List[str]] = None,
    timeout: int = 30
) -> str:
    """Create GitHub issue using gh CLI.

    Args:
        title: Issue title
        body: Issue body (markdown)
        labels: Issue labels (default: None)
        timeout: Command timeout in seconds

    Returns:
        Issue URL

    Raises:
        subprocess.CalledProcessError: If gh command fails
        RuntimeError: If gh CLI not installed

    Example:
        >>> url = create_github_issue(
        ...     "Bug: Login fails",
        ...     "Login button doesn't work",
        ...     labels=["bug", "p1"]
        ... )
    """
    # Build gh command (argument array)
    cmd = ["gh", "issue", "create", "--title", title, "--body", body]

    if labels:
        for label in labels:
            cmd.extend(["--label", label])

    # Execute safely
    result = subprocess.run(
        cmd,
        capture_output=True,
        text=True,
        timeout=timeout,
        check=True,
        shell=False
    )

    # Extract URL from output
    return result.stdout.strip()

import time
from typing import Callable, TypeVar, Any

T = TypeVar('T')

def retry_with_backoff(
    func: Callable[..., T],
    *,
    max_attempts: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0
) -> T:
    """Retry function with exponential backoff.

    Args:
        func: Function to retry
        max_attempts: Maximum retry attempts
        base_delay: Initial delay in seconds
        max_delay: Maximum delay in seconds

    Returns:
        Function result

    Raises:
        Exception: Last exception if all retries fail

    Example:
        >>> result = retry_with_backoff(
        ...     lambda: api_call(),
        ...     max_attempts=5,
        ...     base_delay=2.0
        ... )
    """
    last_exception = None

    for attempt in range(max_attempts):
        try:
            return func()
        except Exception as e:
            last_exception = e

            if attempt < max_attempts - 1:
                # Exponential backoff: 1s, 2s, 4s, 8s, ...
                delay = min(base_delay * (2 ** attempt), max_delay)
                time.sleep(delay)

    raise last_exception

import os
from typing import Optional

def get_github_token() -> str:
    """Get GitHub token from environment.

    Returns:
        GitHub personal access token

    Raises:
        RuntimeError: If token not found

    Security:
        - Environment Variables: Never hardcode tokens
        - Validation: Check token format
        - No Logging: Never log credentials
    """
    token = os.getenv("GITHUB_TOKEN")

    if not token:
        raise RuntimeError(
            "GITHUB_TOKEN not found in environment\n"
            "Set with: export GITHUB_TOKEN=your_token\n"
            "Or add to .env file"
        )

    # Validate token format (basic check)
    if not token.startswith("ghp_") and not token.startswith("github_pat_"):
        raise ValueError("Invalid GitHub token format")

    return token

import time
from datetime import datetime, timedelta

class RateLimiter:
    """Simple rate limiter for API calls.

    Attributes:
        max_calls: Maximum calls per window
        window_seconds: Time window in seconds
    """

    def __init__(self, max_calls: int, window_seconds: int):
        self.max_calls = max_calls
        self.window_seconds = window_seconds
        self.calls = []

    def wait_if_needed(self) -> None:
        """Wait if rate limit would be exceeded."""
        now = datetime.now()
        cutoff = now - timedelta(seconds=self.window_seconds)

        # Remove old calls outside window
        self.calls = [c for c in self.calls if c > cutoff]

        # Wait if at limit
        if len(self.calls) >= self.max_calls:
            oldest = self.calls[0]
            wait_until = oldest + timedelta(seconds=self.window_seconds)
            wait_seconds = (wait_until - now).total_seconds()

            if wait_seconds > 0:
                time.sleep(wait_seconds)

            # Retry removal after wait
            self.calls = [c for c in self.calls if c > cutoff]

        # Record this call
        self.calls.append(now)

name	api-integration-patterns
version	1.0.0
type	knowledge
description	API integration patterns for autonomous-dev including subprocess safety, GitHub CLI integration, retry logic, authentication, rate limiting, and timeout handling. Use when integrating external APIs or CLI tools.
keywords	api, subprocess, github, gh cli, retry, authentication, rate limiting, timeout, command injection, CWE-78, integration
auto_activate	true
allowed-tools	["Read"]

name	api-integration-patterns
version	1.0.0
type	knowledge
description	API integration patterns for autonomous-dev including subprocess safety, GitHub CLI integration, retry logic, authentication, rate limiting, and timeout handling. Use when integrating external APIs or CLI tools.
keywords	api, subprocess, github, gh cli, retry, authentication, rate limiting, timeout, command injection, CWE-78, integration
auto_activate	true
allowed-tools	["Read"]

api-integration-patterns

Installationsoptionen

Quelldateien prüfen