### Health Server Example Usage Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/health.md Example demonstrating how to start the health server, signal readiness, and handle shutdown. ```python from threading import Event from pr_generator.health import start_health_server stop = Event() server, ready_event = start_health_server(port=8080, stop_event=stop) # Later in main loop: # After first successful scan cycle ready_event.set() # On shutdown signal: stop.set() # Server now returns 503 for /livez and /healthz ``` -------------------------------- ### Multi-Provider Setup Configuration Source: https://github.com/devops-ia/pr-generator/blob/main/llms.txt Configuration example for setting up multiple GitHub and Bitbucket providers, including different organizations/workspaces and authentication methods. Each provider requires a unique token environment variable. ```yaml # All providers of the same type share one secret per type. # Each provider MUST use a unique token_env to avoid credential collisions. providers: github-acme: type: github # required for non-standard key names enabled: true auth_method: app owner: acme-org repo: backend app_id: "111" private_key_path: /secrets/acme-app.pem github-skunkworks: type: github enabled: true auth_method: pat owner: skunkworks-org repo: platform token_env: SKUNKWORKS_GITHUB_TOKEN # must be unique across PAT providers bitbucket-eu: type: bitbucket enabled: true workspace: eu-workspace repo_slug: my-repo token_env: BITBUCKET_EU_TOKEN # must be unique across Bitbucket providers bitbucket-us: type: bitbucket enabled: true workspace: us-workspace repo_slug: my-repo token_env: BITBUCKET_US_TOKEN # different env var — required rules: - pattern: "feature/.*" destinations: github-acme: main github-skunkworks: develop bitbucket-eu: develop bitbucket-us: develop ``` -------------------------------- ### Install and Run PR Generator Source: https://github.com/devops-ia/pr-generator/blob/main/README.md Install the PR generator using pip and run it by pointing to your configuration file. Alternatively, use Docker for a quick setup. ```bash # Install pip install -e . # Point to your config file and run CONFIG_PATH=./config.yaml pr-generator ``` ```bash docker run --rm \ -v "$(pwd)/config.yaml:/etc/pr-generator/config.yaml:ro" \ ghcr.io/devops-ia/pr-generator:latest ``` -------------------------------- ### Configuration Example (GitHub App + Bitbucket) Source: https://github.com/devops-ia/pr-generator/blob/main/index.html Example config.yaml demonstrating how to configure both GitHub App and Bitbucket providers, along with rules for generating PRs. Credentials should be managed via environment variables or secrets. ```yaml scan_frequency: 60 # seconds between scan cycles dry_run: false # set true to simulate without creating PRs providers: github: # name is free-form; type inferred from key enabled: true owner: my-org repo: my-app auth_method: app # "app" (PEM) or "pat" (token) app_id: "123456" # from env: GITHUB_APP_ID installation_id: "78901234" private_key_path: /secrets/github.pem # or use GITHUB_APP_PRIVATE_KEY env bitbucket: enabled: true workspace: my-workspace repo_slug: my-app token_env: BITBUCKET_TOKEN # env var holding the access token rules: - name: image-updater-prs branch_pattern: "^image-updater/.*" destinations: - provider: github base: main - provider: bitbucket base: main ``` -------------------------------- ### Install from PyPI Source: https://github.com/devops-ia/pr-generator/blob/main/index.html Install the pr-generator package using pip and run it with a configuration file. ```bash $ pip install pr-generator $ pr-generator --config /etc/pr-generator/config.yaml ``` -------------------------------- ### Get Installation Token Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/providers-github.md Returns a cached installation token or requests a new one from the API. Installation tokens are short-lived, approximately one hour. The token is refreshed if it will expire within the next 30 seconds. Uses JWT authentication for API calls. ```python def _get_installation_token(self) -> str: ``` ```python # Token is automatically refreshed as needed by get_branches, check_existing_pr, etc. ``` -------------------------------- ### GitHub Provider with App Authentication Source: https://github.com/devops-ia/pr-generator/blob/main/llms.txt Instantiate and use the GitHubProvider with GitHub App authentication. This example shows how to configure the provider with app details and then perform actions like getting branches, checking for existing PRs, and creating new ones. ```python from pr_generator.models import ProviderConfig from pr_generator.providers.github import GitHubProvider, GitHubError # GitHub App config = ProviderConfig( name="github", enabled=True, type="github", owner="my-org", repo="my-repo", auth_method="app", app_id="123456", installation_id="78901234", # optional — auto-resolved if empty private_key="-----BEGIN RSA PRIVATE KEY-----\n...", timeout=30.0, ) provider = GitHubProvider(config) try: branches = provider.get_branches() # ["main", "develop", "feature/x"] exists = provider.check_existing_pr("feature/x", "main") # True / False if not exists: provider.create_pull_request("feature/x", "main") provider.reset_cycle_cache() # call at start of each cycle except GitHubError as exc: print(f"error {exc.status_code}: {exc}") ``` -------------------------------- ### Start Health Server and Simulate Lifecycle Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/health.md This Python snippet demonstrates how to start the health server and simulate the application's lifecycle, including readiness events and shutdown. ```python from threading import Event from pr_generator.health import start_health_server import time stop = Event() # Start health server server, ready_event = start_health_server(port=8080, stop_event=stop) # Simulate scanning time.sleep(2) # After first cycle ready_event.set() print("Ready") # Later, on shutdown signal (SIGTERM, SIGINT) stop.set() print("Shutting down") ``` -------------------------------- ### Initialize Logging in Python Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/logging_config.md Example of how to set up logging using the `setup_logging` function. It shows early bootstrap and reconfiguration after loading configuration. ```python from pr_generator.logging_config import setup_logging # At startup (before config is loaded) setup_logging("INFO") # After config is loaded (with user-specified settings) setup_logging(config.log_level, json_format=(config.log_format == "json")) ``` -------------------------------- ### Initializing BitbucketProvider Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/providers-bitbucket.md Example of how to initialize the BitbucketProvider with a ProviderConfig object. Ensure all required fields like workspace, repo_slug, and token are provided. ```python from pr_generator.providers.bitbucket import BitbucketProvider from pr_generator.models import ProviderConfig config = ProviderConfig( name="bitbucket", enabled=True, type="bitbucket", workspace="my-workspace", repo_slug="my-repo", token="", close_source_branch=True, ) provider = BitbucketProvider(config) ``` -------------------------------- ### Example HeadersFactory Implementation Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/http_client.md An example implementation of the HeadersFactory. This function generates headers including a dynamically fetched JWT for authorization and sets the Accept header. ```python def headers_factory() -> dict: return { "Authorization": f"Bearer {get_fresh_jwt()}", "Accept": "application/vnd.github+json", } ``` -------------------------------- ### Install and Run pr-generator Source: https://github.com/devops-ia/pr-generator/blob/main/llms.txt Install the tool from source using pip and run it with default or custom configuration paths. Includes a command to check the version. ```bash pip install -e . pr-generator CONFIG_PATH=./config.yaml pr-generator pr-generator --version ``` -------------------------------- ### Python Virtual Environment Setup Source: https://github.com/devops-ia/pr-generator/blob/main/README.md Commands to create and activate a Python virtual environment. ```bash # Create and activate a virtual environment python -m venv .venv source .venv/bin/activate ``` -------------------------------- ### Start and Manage Health Server Source: https://github.com/devops-ia/pr-generator/blob/main/llms.txt Illustrates how to start a health server on a specified port and manage its lifecycle using threading Events. It shows how to signal readiness and initiate a graceful shutdown. ```python from threading import Event from pr_generator.health import start_health_server stop_event = Event() server, ready_event = start_health_server(port=8080, stop_event=stop_event) # After first successful scan cycle: ready_event.set() # /readyz starts returning 200 # Graceful shutdown: stop_event.set() # /livez starts returning 503 ``` -------------------------------- ### FileNotFoundError Example Source: https://github.com/devops-ia/pr-generator/blob/main/README.md Example of a FileNotFoundError when the private key path is incorrect. ```text FileNotFoundError: [Core] private_key_path '/secrets/github-app.pem' does not exist. ``` -------------------------------- ### Structured Logging Format Examples Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/logging_config.md Examples of structured log messages using a key=value pattern. This format ensures consistency and aids parsing. ```text [Core] Step: load_config action=start source=file path=/etc/pr-generator/config.yaml [Core] Step: load_config action=end source=file providers=['github', 'bitbucket'] rules=3 [GitHub] Step: get_branches action=start [GitHub] Step: get_branches action=fetch page=1 [GitHub] Step: get_branches action=end total=42 [GitHub] Step: check_existing_pr action=start source=feature/x dest=main [GitHub] Step: check_existing_pr action=cache_hit source=feature/x dest=main [GitHub] Step: create_pull_request action=start source=feature/x dest=main [GitHub] Step: create_pull_request action=end source=feature/x dest=main status=created number=123 [Core] Step: process_rule action=start cycle_id=1 pattern=feature/.* dest=main [Core] Step: process_rule action=end cycle_id=1 pattern=feature/.* dest=main processed=3 created=2 dry_run=0 skipped=1 errors=0 ``` -------------------------------- ### ScanRule Example Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/models.md Creates a ScanRule to match branches starting with 'feature/'. This rule defines destination branches for GitHub and Bitbucket. ```python from pr_generator.models import ScanRule import re rule = ScanRule( pattern="feature/.*", compiled=re.compile("feature/.*"), destinations={ "github": "develop", "bitbucket": "develop", } ) # This rule matches branches like: # - feature/user-login # - feature/api-v2 # But not: # - hotfix/feature/x (doesn't start with "feature/") # - develop (excluded: destination branch) ``` -------------------------------- ### Install Development Dependencies Source: https://github.com/devops-ia/pr-generator/blob/main/README.md Commands to install the package in editable mode with development extras and pytest. ```bash # Install the package in editable mode with dev extras pip install -e . pip install pytest ``` -------------------------------- ### Full Lifecycle Example with BitbucketProvider Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/providers-bitbucket.md Demonstrates the complete lifecycle of using the Bitbucket provider, including configuration, fetching branches, and creating pull requests. Ensure you have the necessary access token and repository details. ```python from pr_generator.providers.bitbucket import BitbucketProvider from pr_generator.models import ProviderConfig config = ProviderConfig( name="bitbucket", enabled=True, type="bitbucket", workspace="my-workspace", repo_slug="my-repo", token="", close_source_branch=True, ) provider = BitbucketProvider(config) # Phase 1: Fetch branches branches = provider.get_branches() print(f"Found {len(branches)} branches") # Phase 2: Process rules provider.reset_cycle_cache() matched = [b for b in branches if b.startswith("feature/")] for branch in matched: if not provider.check_existing_pr(branch, "develop"): try: provider.create_pull_request(branch, "develop") print(f"Created PR: {branch} → develop") except BitbucketError as e: print(f"Error: {e}") ``` -------------------------------- ### Example Provider Configuration Source: https://github.com/devops-ia/pr-generator/blob/main/index.html Demonstrates how to assign unique environment variable names for distinct providers of the same type to avoid 'duplicate token_env' errors. ```yaml providers: bitbucket: token_env: BITBUCKET_TOKEN bitbucket-org2: type: bitbucket token_env: BITBUCKET_ORG2_TOKEN # must differ ``` -------------------------------- ### PR Generator Configuration Example Source: https://github.com/devops-ia/pr-generator/blob/main/README.md Example YAML configuration file for PR generator, detailing settings for scan frequency, logging, providers (GitHub, Bitbucket), and rules. ```yaml # config.yaml # How often (seconds) to scan for new branches. scan_frequency: 300 # default: 300 # Logging level: DEBUG | INFO | WARNING | ERROR log_level: INFO # default: INFO # Log format: "text" (human-readable) or "json" (structured, for log aggregators) log_format: text # default: text # When true, PRs are logged but never actually created. dry_run: false # default: false # Port for the built-in health server. health_port: 8080 # default: 8080 providers: github: enabled: true owner: my-org repo: my-repo app_id: "123456" installation_id: "78901234" # optional — auto-resolved if omitted private_key_path: /secrets/github-app.pem # path to PEM file # Alternative: set GITHUB_APP_PRIVATE_KEY env var (plain PEM or base64-encoded) timeout: 30 # HTTP timeout in seconds bitbucket: enabled: true workspace: my-workspace repo_slug: my-repo token_env: BITBUCKET_TOKEN # name of the env var that holds the token close_source_branch: true # delete source branch after merge (default: true) timeout: 30 rules: - pattern: "feature/.*" # Python regex matched against branch names destinations: github: main bitbucket: develop - pattern: "release/.*" destinations: github: main - pattern: ".*-hotfix-.*" destinations: bitbucket: master ``` -------------------------------- ### Example Usage of _process_rule Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/scanner.md Demonstrates how to use the _process_rule function with GitHubProvider, a list of branches, and a ScanRule. Shows how to interpret the results. ```python from pr_generator.providers.github import GitHubProvider from pr_generator.scanner import _process_rule from pr_generator.models import ScanRule import re provider = GitHubProvider(config.providers["github"]) branches = ["feature/x", "feature/y", "hotfix/z"] rule = ScanRule( pattern="feature/.*", compiled=re.compile("feature/.*"), destinations={"github": "develop"} ) result = _process_rule( provider=provider, branches=branches, rule=rule, dest_branch="develop", dry_run=False, cycle_id=1, ) print(f"Processed {result.processed}, Created {result.created}, Errors {result.errors}") # Output: Processed 2, Created 2, Errors 0 # (feature/x and feature/y matched; hotfix/z didn't) ``` -------------------------------- ### Text Format Log Output Example Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/logging_config.md Example of standard text-based log output. This format is human-readable and includes timestamps and log levels. ```text 2025-02-20T14:30:45 INFO pr_generator [Core] Step: startup action=end detail=Configuration loaded 2025-02-20T14:30:45 INFO pr_generator [Core] Active providers: github, bitbucket 2025-02-20T14:30:45 INFO pr_generator [Core] Rules configured: 3 2025-02-20T14:30:45 INFO pr_generator [Core] Rule: pattern=feature/.* destinations={'github': 'main', 'bitbucket': 'develop'} 2025-02-20T14:30:46 INFO pr_generator.scanner [Core] Step: scan_cycle action=start cycle_id=1 rules=3 providers=['github', 'bitbucket'] 2025-02-20T14:30:46 INFO pr_generator.providers.github [github] Step: get_branches action=start 2025-02-20T14:30:47 INFO pr_generator.providers.github [github] Step: get_branches action=end total=15 ``` -------------------------------- ### ProviderConfig Example Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/models.md Instantiates a ProviderConfig for GitHub with App authentication. This configuration is passed to provider implementations. ```python from pr_generator.models import ProviderConfig github_config = ProviderConfig( name="github", enabled=True, type="github", owner="my-org", repo="my-repo", app_id="123456", private_key="""-----BEGIN RSA PRIVATE KEY----- ...""", ) provider = GitHubProvider(github_config) ``` -------------------------------- ### Example Usage of scan_cycle Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/scanner.md Demonstrates how to load configuration, instantiate providers (GitHub and Bitbucket), and run a scan cycle using the scan_cycle function. It also shows how to aggregate and print statistics from the cycle result. ```python from pr_generator.config import load_config from pr_generator.providers.github import GitHubProvider from pr_generator.providers.bitbucket import BitbucketProvider from pr_generator.scanner import scan_cycle config = load_config() # Instantiate providers providers = { "github": GitHubProvider(config.providers["github"]), "bitbucket": BitbucketProvider(config.providers["bitbucket"]), } # Run one cycle result = scan_cycle(config, providers, cycle_id=1) # Aggregate statistics total_processed = sum(r.processed for r in result.rule_results) total_created = sum(r.created for r in result.rule_results) print(f"Cycle {result.cycle_id}: processed {total_processed}, created {total_created}") ``` -------------------------------- ### Example: Checking Readyz Endpoint Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/endpoints.md Demonstrates checking the /readyz endpoint using curl. Shows expected output before the first scan, after completion, and during shutdown. ```bash # During startup (before first scan): curl http://localhost:8080/readyz # Output: not ready (HTTP 503) # After first scan completes: curl http://localhost:8080/readyz # Output: ready (HTTP 200) # During shutdown: curl http://localhost:8080/readyz # Output: not ready (HTTP 503) ``` -------------------------------- ### Full GitHub Provider Lifecycle Example Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/providers-github.md Demonstrates the full lifecycle of using the GitHub provider, from fetching branches to creating pull requests. Includes error handling for PR creation. ```python from pr_generator.providers.github import GitHubProvider from pr_generator.models import ProviderConfig config = ProviderConfig( name="github", enabled=True, type="github", owner="my-org", repo="my-repo", app_id="123456", private_key="""-----BEGIN RSA PRIVATE KEY----- ...""", ) provider = GitHubProvider(config) # Phase 1: Fetch branches branches = provider.get_branches() print(f"Found {len(branches)} branches") # Phase 2: Process rules provider.reset_cycle_cache() matched = [b for b in branches if b.startswith("feature/")] for branch in matched: if not provider.check_existing_pr(branch, "main"): try: provider.create_pull_request(branch, "main") print(f"Created PR: {branch} → main") except GitHubError as e: print(f"Error: {e}") ``` -------------------------------- ### Start Health Server Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/health.md Starts the health HTTP server in a daemon thread. Returns the server instance and a ready event that is set after the first successful cycle. ```python def start_health_server( port: int, stop_event: Event ) -> tuple[ThreadingHTTPServer, Event]: """Start the health HTTP server in a daemon thread. Returns: (server, ready_event) — set ready_event after the first successful cycle. """ ``` -------------------------------- ### Install with Helm Chart Source: https://github.com/devops-ia/pr-generator/blob/main/index.html Deploy pr-generator on Kubernetes using the official Helm chart. This command sets up the necessary repository, installs the chart, and configures basic GitHub and secret parameters. ```bash $ helm repo add helm-pr-generator https://devops-ia.github.io/helm-pr-generator $ helm repo update $ helm install pr-generator helm-pr-generator/pr-generator \ --namespace pr-generator --create-namespace \ --set config.providers.github.owner=my-org \ --set config.providers.github.repo=my-repo \ --set config.providers.github.appId="123456" \ --set secrets.github.privateKey="$(cat github-app.pem)" ``` -------------------------------- ### JSON Log Output Example Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/logging_config.md Example of newline-delimited JSON log lines. This format is ideal for log aggregators. ```json {"timestamp": "2025-02-20 14:30:45,123", "level": "INFO", "logger": "pr_generator.scanner", "message": "[Core] Step: scan_cycle action=start cycle_id=1"} {"timestamp": "2025-02-20 14:30:46,234", "level": "DEBUG", "logger": "pr_generator.providers.github", "message": "[GitHub] Step: get_branches action=fetch page=1"} {"timestamp": "2025-02-20 14:30:47,345", "level": "INFO", "logger": "pr_generator.scanner", "message": "[Core] Step: scan_cycle action=end cycle_id=1 processed=3 created=1"} ``` -------------------------------- ### Generate HTTP Headers (Installation Token) Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/providers-github.md Generates HTTP headers for API requests using an installation token. Includes the 'Authorization' header with a Bearer token and the 'Accept' header for GitHub API v3. ```python { "Authorization": f"Bearer {installation_token}", "Accept": "application/vnd.github+json", } ``` -------------------------------- ### ValueError: duplicate tokenEnv Example Source: https://github.com/devops-ia/pr-generator/blob/main/README.md Example of a ValueError indicating duplicate tokenEnv usage among providers. ```text ValueError: [Core] Providers 'bb-eu' and 'bb-us' both use tokenEnv 'BITBUCKET_TOKEN'. ``` -------------------------------- ### Example ShouldRetry Implementation Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/http_client.md An example implementation of the ShouldRetry predicate. This function retries on network errors or specific HTTP status codes like 5xx and 429 (rate limiting). ```python def should_retry(status_code: int | None, exc: Exception | None) -> bool: if exc is not None: # Retry on network errors return True # Retry on 5xx and rate limit (429) return bool(status_code and (500 <= status_code < 600 or status_code == 429)) ``` -------------------------------- ### Text Log Format Example Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/configuration.md An example of the human-readable text log format, showing timestamp, log level, logger name, and message details. ```text 2025-02-20T14:30:45 INFO pr_generator.scanner [Core] Step: scan_cycle action=start cycle_id=1 rules=2 providers=['github'] ``` -------------------------------- ### Cache Reset Example Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/scanner.md Illustrates the per-cycle cache reset mechanism for provider instances to ensure fresh data for branch and PR checks. ```python for prov in providers.values(): prov.reset_cycle_cache() ``` -------------------------------- ### Example: Checking Livez Endpoint Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/endpoints.md Demonstrates how to check the /livez endpoint using curl. Shows expected output during normal operation and shutdown. ```bash curl http://localhost:8080/livez # Output: live (exit code 0) # During shutdown: # Output: shutting down (exit code 1, HTTP 503) ``` -------------------------------- ### Setup Logging with Text or JSON Format Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/logging_config.md Configure the root logger for text or JSON output. Use text format for human readability and JSON format for log aggregators. Invalid level names default to INFO. ```python from pr_generator.logging_config import setup_logging # Text format (default) setup_logging("INFO") # Output: # 2025-02-20T14:30:45 INFO pr_generator.scanner [Core] Step: scan_cycle... # JSON format (for log aggregators) setup_logging("DEBUG", json_format=True) # Output: # {"timestamp": "2025-02-20 14:30:45,123", "level": "DEBUG", "logger": "pr_generator.scanner", "message": "[Core] Step: scan_cycle..."} ``` -------------------------------- ### Run pr-generator with GitHub PAT Source: https://github.com/devops-ia/pr-generator/blob/main/llms.txt Example command to run pr-generator using a GitHub PAT, setting the token via an environment variable and specifying the configuration path. ```bash export GITHUB_TOKEN=ghp_xxxxxxxxxxxxxxxxxxxx CONFIG_PATH=./config.yaml pr-generator ``` -------------------------------- ### GET /readyz Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/endpoints.md Checks if the application is ready to serve traffic. It returns 503 until the first successful scan cycle completes, after which it returns 200. It also returns 503 during shutdown. ```APIDOC ## GET /readyz ### Description Checks if the application is ready to serve traffic. This endpoint returns a 503 status code until the initial scan cycle is successfully completed. Once ready, it returns 200. During a graceful shutdown, it also returns 503. ### Method GET ### Endpoint /readyz ### Response #### Success Response (200) - **Status**: Application is ready (first scan cycle completed and not shutting down) ``` ready ``` #### Error Response (503) - **Status**: Not ready yet (before first cycle or during shutdown) ``` not ready ``` ### Use case Kubernetes readiness probe. Prevents traffic routing until the application has completed its first full scan cycle, ensuring all providers are initialized and config is valid. ### Request Example ```bash # During startup (before first scan): curl http://localhost:8080/readyz # Output: not ready (HTTP 503) # After first scan completes: curl http://localhost:8080/readyz # Output: ready (HTTP 200) # During shutdown: curl http://localhost:8080/readyz # Output: not ready (HTTP 503) ``` ``` -------------------------------- ### Creating a RuleResult Instance Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/models.md Example of how a `RuleResult` object might be populated after processing a rule. This is typically done internally within the scanner. ```python # In scanner._process_rule (after processing): result = RuleResult( rule_pattern="feature/.*", provider="github", destination="develop", processed=5, created=3, skipped_existing=2, simulated=0, errors=0, ) ``` -------------------------------- ### start_health_server Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/health.md Starts the health HTTP server in a daemon thread. It listens on a specified port and handles liveness, readiness, and other health check endpoints. ```APIDOC ## start_health_server() ### Description Starts the health HTTP server in a daemon thread. This server is designed for Kubernetes liveness, readiness, and startup probes. ### Method Python function call ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Parameters - **port** (int) - Required - Port to bind to (usually from `config.health_port`) - **stop_event** (Event) - Required - Threading event signaling shutdown (set by main loop) ### Returns - **ThreadingHTTPServer**: HTTP server instance (for cleanup if needed) - **Event**: `ready_event` to signal after first cycle completes ### Endpoints Handled by Server: - `GET /livez` → 200 "live" (or 503 "shutting down") - `GET /healthz` → 200 "live" (or 503 "shutting down") - `GET /readyz` → 200 "ready" (or 503 "not ready") - `GET ` → 404 "not found" ### Example ```python from threading import Event from pr_generator.health import start_health_server stop = Event() server, ready_event = start_health_server(port=8080, stop_event=stop) # Later in main loop: # After first successful scan cycle ready_event.set() # On shutdown signal: stop.set() ``` ``` -------------------------------- ### Docker Compose Configuration Source: https://github.com/devops-ia/pr-generator/blob/main/index.html Example docker-compose.yml for running pr-generator. Ensure to mount a configuration file and provide GitHub App credentials via environment variables or secrets. ```yaml services: pr-generator: image: ghcr.io/devops-ia/pr-generator:latest restart: unless-stopped volumes: - ./config.yaml:/etc/pr-generator/config.yaml:ro environment: GITHUB_APP_ID: "123456" GITHUB_INSTALLATION_ID: "78901234" secrets: - github_pem healthcheck: test: ["CMD", "wget", "-qO-", "http://localhost:8080/health"] interval: 30s retries: 3 secrets: github_pem: file: ./github-app.pem ``` -------------------------------- ### Kubernetes Health Probe Configuration Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/endpoints.md Example Kubernetes container configuration specifying liveness, readiness, and startup probes using the HTTP endpoints. ```yaml spec: containers: - name: pr-generator image: pr-generator:latest ports: - containerPort: 8080 name: health livenessProbe: httpGet: path: /livez port: 8080 initialDelaySeconds: 10 periodSeconds: 30 timeoutSeconds: 5 failureThreshold: 3 readinessProbe: httpGet: path: /readyz port: 8080 initialDelaySeconds: 5 periodSeconds: 10 timeoutSeconds: 5 failureThreshold: 3 startupProbe: httpGet: path: /readyz port: 8080 initialDelaySeconds: 0 periodSeconds: 5 timeoutSeconds: 5 failureThreshold: 30 ``` -------------------------------- ### Python Exception Constructor Example Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/http_client.md Illustrates how to construct provider-specific exceptions, which must accept a message string and an optional status code. This is crucial for consistent error handling within the retry mechanism. ```python exception = exception_cls(message: str, status_code: int | None) # Example: GitHubError("API error 500", 500) or GitHubError("Connection timeout", None) ``` -------------------------------- ### Create Pull Request Example Usage Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/providers-base.md Demonstrates how to use the `create_pull_request` method within a try-except block to handle potential API errors, such as `GitHubError`. ```python try: provider.create_pull_request("feature/x", "develop") print("PR created successfully") except GitHubError as e: print(f"Failed to create PR: {e}") ``` -------------------------------- ### JSON Format Log Output Comparison Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/logging_config.md Example of JSON formatted log output, demonstrating structured fields and ISO timestamps. This is suitable for automated log processing. ```json {"timestamp": "2025-02-20 14:30:45,123", "level": "INFO", "logger": "pr_generator", "message": "[Core] Step: startup action=end detail=Configuration loaded"} {"timestamp": "2025-02-20 14:30:45,124", "level": "INFO", "logger": "pr_generator", "message": "[Core] Active providers: github, bitbucket"} {"timestamp": "2025-02-20 14:30:45,125", "level": "INFO", "logger": "pr_generator", "message": "[Core] Rules configured: 3"} {"timestamp": "2025-02-20 14:30:46,234", "level": "INFO", "logger": "pr_generator.scanner", "message": "[Core] Step: scan_cycle action=start cycle_id=1 rules=3 providers=['github', 'bitbucket']"} {"timestamp": "2025-02-20 14:30:46,235", "level": "INFO", "logger": "pr_generator.providers.github", "message": "[github] Step: get_branches action=start"} {"timestamp": "2025-02-20 14:30:47,345", "level": "INFO", "logger": "pr_generator.providers.github", "message": "[github] Step: get_branches action=end total=15"} ``` -------------------------------- ### Kubernetes Liveness Probe Configuration Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/health.md Example YAML configuration for a Kubernetes liveness probe using the /livez HTTP endpoint. Specifies initial delay, period, and failure threshold. ```yaml livenessProbe: httpGet: path: /livez port: 8080 initialDelaySeconds: 10 periodSeconds: 30 failureThreshold: 3 ``` -------------------------------- ### Define ScanRule Data Model Source: https://github.com/devops-ia/pr-generator/blob/main/llms.txt Illustrates the ScanRule data model, which defines patterns for branch names and maps them to destination branches for different providers. Includes an example using regular expressions. ```python from pr_generator.models import ScanRule import re # ScanRule rule = ScanRule( pattern="feature/.*", compiled=re.compile("feature/.*", re.IGNORECASE), destinations={"github": "main", "bitbucket": "develop"}, ) ``` -------------------------------- ### Bitbucket Provider Initialization and Usage Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/providers-base.md Demonstrates initializing a `BitbucketProvider` with configuration and calling its core methods like `get_branches`, `check_existing_pr`, `create_pull_request`, and `reset_cycle_cache`. ```python from pr_generator.providers.bitbucket import BitbucketProvider from pr_generator.models import ProviderConfig config = ProviderConfig( name="bitbucket", enabled=True, type="bitbucket", workspace="my-workspace", repo_slug="my-repo", token="", ) provider = BitbucketProvider(config) # All methods are available assert provider.name == "bitbucket" branches = provider.get_branches() exists = provider.check_existing_pr("feature/x", "develop") if not exists: provider.create_pull_request("feature/x", "develop") provider.reset_cycle_cache() ``` -------------------------------- ### Initialize and Use Scan Cycle Source: https://github.com/devops-ia/pr-generator/blob/main/llms.txt Shows how to load configuration, initialize GitHub and Bitbucket providers, and run a scan cycle to process rules and generate pull requests. It iterates through the results, printing details about processed, created, and skipped pull requests. ```python from pr_generator.config import load_config from pr_generator.scanner import scan_cycle from pr_generator.providers.github import GitHubProvider from pr_generator.providers.bitbucket import BitbucketProvider config = load_config() providers = {} for name, pconf in config.providers.items(): if pconf.type == "github": providers[name] = GitHubProvider(pconf) elif pconf.type == "bitbucket": providers[name] = BitbucketProvider(pconf) result = scan_cycle(config, providers, cycle_id=1) for r in result.rule_results: print(f"pattern={r.rule_pattern} provider={r.provider} dest={r.destination}") print(f" processed={r.processed} created={r.created} " f"skipped={r.skipped_existing} dry_run={r.simulated} errors={r.errors}") ``` -------------------------------- ### Health Handler GET Request Handling Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/health.md Handles GET requests by dispatching based on the request path to implement liveness and readiness checks. ```python def do_GET(self) -> None: ``` -------------------------------- ### GitHubProvider Initialization Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/providers-github.md Demonstrates initializing the GitHubProvider with a ProviderConfig object, including GitHub App credentials. ```python from pr_generator.providers.github import GitHubProvider from pr_generator.models import ProviderConfig config = ProviderConfig( name="github", enabled=True, type="github", owner="my-org", repo="my-repo", app_id="123456", private_key="-----BEGIN RSA PRIVATE KEY----- ...", ) provider = GitHubProvider(config) ``` -------------------------------- ### JSON Log Format Example Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/configuration.md An example of the structured JSON log format, suitable for log aggregators. Includes timestamp, level, logger, and message. ```json {"timestamp": "2025-02-20 14:30:45,123", "level": "INFO", "logger": "pr_generator.scanner", "message": "[Core] Step: scan_cycle action=start cycle_id=1 rules=2 providers=['github']"} ``` -------------------------------- ### GitHub Provider Initialization and Usage Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/providers-base.md Demonstrates initializing a `GitHubProvider` with configuration and calling its core methods like `get_branches`, `check_existing_pr`, `create_pull_request`, and `reset_cycle_cache`. ```python from pr_generator.providers.github import GitHubProvider from pr_generator.models import ProviderConfig config = ProviderConfig( name="github", enabled=True, type="github", owner="my-org", repo="my-repo", app_id="123456", private_key="-----BEGIN RSA PRIVATE KEY-----\n...", ) provider = GitHubProvider(config) # All methods are available assert provider.name == "github" branches = provider.get_branches() exists = provider.check_existing_pr("feature/x", "main") if not exists: provider.create_pull_request("feature/x", "main") provider.reset_cycle_cache() ``` -------------------------------- ### BitbucketProvider Initialization Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/providers-bitbucket.md Initializes the BitbucketProvider with necessary configuration details for connecting to a Bitbucket Cloud repository. ```APIDOC ## BitbucketProvider ```python class BitbucketProvider: """Bitbucket Cloud provider. Receives all configuration via constructor — no module-level env-var reads. """ ``` ### __init__(self, config: ProviderConfig) **Parameters**: | Parameter | Type | Description | |-----------|------|-------------| | `config` | ProviderConfig | Configuration object with all necessary credentials | **Initialization**: 1. Stores configuration fields (workspace, repo_slug, token, close_source_branch) 2. Constructs API endpoint root URL 3. Initializes per-cycle PR-existence cache **Per-Cycle Cache** (cleared each cycle via `reset_cycle_cache`): - `_pr_cache`: PR existence lookups (source, destination) → bool **Example**: ```python from pr_generator.providers.bitbucket import BitbucketProvider from pr_generator.models import ProviderConfig config = ProviderConfig( name="bitbucket", enabled=True, type="bitbucket", workspace="my-workspace", repo_slug="my-repo", token="", close_source_branch=True, ) provider = BitbucketProvider(config) ``` ``` -------------------------------- ### setup_logging Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/logging_config.md Configures the root logger for the application. It allows setting the logging level and choosing between a human-readable text format or a structured JSON format for log output. ```APIDOC ## setup_logging() ### Description Configures the root logger with a specified level and format. Supports both text and JSON output. ### Method Python Function ### Signature `setup_logging(level: str, json_format: bool = False) -> None` ### Parameters #### Arguments - **level** (str) - Required - Logging level string, e.g., "INFO", "DEBUG". Valid levels are "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL". Invalid names default to "INFO". - **json_format** (bool) - Optional - If True, emits structured JSON lines. Defaults to False for human-readable text format. ### Returns `None` ### Behavior 1. Gets the root logger. 2. Sets the logging level. 3. Creates a `StreamHandler`. 4. Attaches a formatter based on `json_format` (Text or JSON). 5. Replaces root logger handlers. ### Example ```python from pr_generator.logging_config import setup_logging # Text format (default) setup_logging("INFO") # JSON format (for log aggregators) setup_logging("DEBUG", json_format=True) ``` ``` -------------------------------- ### Resolve Installation ID Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/providers-github.md Fetches the GitHub App installation ID from the API if it's not already stored. This automatic resolution helps prevent configuration errors. The resolved ID is cached for future use. Raises RuntimeError if auto-resolution fails. ```python def _resolve_installation_id(self) -> str: ``` -------------------------------- ### Get Branches Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/providers-github.md Fetches a list of all branches in the repository. ```APIDOC ## GET /repos/{owner}/{repo}/branches ### Description Retrieves a list of all branches within the specified repository. ### Method GET ### Endpoint `/repos/{owner}/{repo}/branches` ### Parameters #### Path Parameters - **owner** (string) - Required - The owner of the repository. - **repo** (string) - Required - The name of the repository. ### Response #### Success Response (200) An array of branch objects. #### Response Example ```json [ { "name": "main", "commit": { "sha": "abcdef1234567890" } }, { "name": "develop", "commit": { "sha": "fedcba0987654321" } } ] ``` ``` -------------------------------- ### CSS Quick-start Block Styles Source: https://github.com/devops-ia/pr-generator/blob/main/index.html Styles the quick-start code block, including its background, border, header, and code formatting with syntax highlighting. ```css .quickstart { background: var(--code-bg); border: 1px solid var(--border); border-radius: var(--radius); overflow: hidden; max-width: 640px; margin: 0 auto; text-align: left; } .quickstart-header { display: flex; align-items: center; justify-content: space-between; padding: 10px 16px; background: var(--surface); border-bottom: 1px solid var(--border); font-family: var(--mono); font-size: 0.75rem; color: var(--muted); } .copy-btn { background: none; border: 1px solid var(--border); color: var(--fg-dim); font-family: var(--mono); font-size: 0.7rem; padding: 3px 10px; border-radius: 4px; cursor: pointer; transition: border-color 150ms, color 150ms; } .copy-btn:hover { border-color: var(--accent); color: var(--accent); } .copy-btn.copied { border-color: var(--accent); color: var(--accent); } .quickstart pre { padding: 20px; font-family: var(--mono); font-size: 0.8125rem; line-height: 1.8; overflow-x: auto; } .quickstart pre .prompt { color: var(--accent); user-select: none; } .quickstart pre .cmd { color: #7DD3FC; } .quickstart pre .arg { color: #C084FC; } .quickstart pre .val { color: #FDE68A; } .quickstart pre .comment { color: var(--muted); } ``` -------------------------------- ### Get Branches Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/providers-bitbucket.md Fetches all branch names from the Bitbucket repository, handling pagination automatically. ```APIDOC ### get_branches() ```python def get_branches(self) -> list[str]: """Fetch all branch names (handles pagination).""" ``` **Returns**: `list[str]` — All branch names in the repository **Raises**: `BitbucketError` on API failure **Implementation**: 1. Validates configuration (token, workspace, repo_slug) 2. Paginates through Bitbucket API `/repositories/{workspace}/{repo_slug}/refs/branches` endpoint 3. Extracts branch names from paginated response 4. Returns complete list **Pagination**: Requests `pagelen=100` per page and continues while `next` field present in response. **API Response Format**: ```json { "pagelen": 100, "values": [ {"name": "main", ...}, {"name": "develop", ...}, ... ], "page": 1, "next": "https://api.bitbucket.org/2.0/repositories/.../refs/branches?page=2" } ``` **Example**: ```python branches = provider.get_branches() # Output: ['main', 'develop', 'feature/x', 'feature/y', ...] ``` ``` -------------------------------- ### Handling GitHubError Exceptions Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/providers-github.md Example of how to catch and handle GitHubError, checking the status code for specific actions. ```python try: provider.create_pull_request("feature/x", "main") except GitHubError as e: if e.status_code == 404: print("Repository not found") elif e.status_code and 500 <= e.status_code < 600: print("GitHub server error; will retry") else: print(f"GitHub error: {e}") ``` -------------------------------- ### Loading and Accessing AppConfig Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/models.md Demonstrates how to load the application configuration and access its fields. Ensure the `pr_generator.config` module is available. ```python from pr_generator.config import load_config config = load_config() print(f"Scan every {config.scan_frequency} seconds") print(f"Providers: {list(config.providers.keys())}") print(f"Rules: {len(config.rules)}") for rule in config.rules: print(f" - {rule.pattern} → {rule.destinations}") ``` -------------------------------- ### Handling BitbucketError Exceptions Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/providers-bitbucket.md Example of how to catch and handle BitbucketError, checking the status code for specific error conditions. ```python try: provider.create_pull_request("feature/x", "develop") except BitbucketError as e: if e.status_code == 404: print("Repository not found") elif e.status_code and 500 <= e.status_code < 600: print("Bitbucket server error; will retry") else: print(f"Bitbucket error: {e}") ``` -------------------------------- ### Executing a Scan Cycle and Processing Results Source: https://github.com/devops-ia/pr-generator/blob/main/_autodocs/api-reference/models.md Shows how to initiate a scan cycle using `scan_cycle` and then iterate through the `CycleResult` to display aggregated statistics. Requires `config` and `providers` to be defined. ```python from pr_generator.scanner import scan_cycle # Assuming 'config' and 'providers' are defined and available # result = scan_cycle(config, providers, cycle_id=1) # Example of processing a hypothetical result object: class MockRuleResult: def __init__(self, provider, rule_pattern, processed, created, skipped_existing): self.provider = provider self.rule_pattern = rule_pattern self.processed = processed self.created = created self.skipped_existing = skipped_existing class MockCycleResult: def __init__(self, cycle_id, rule_results): self.cycle_id = cycle_id self.rule_results = rule_results result = MockCycleResult(1, [ MockRuleResult("github", "feature/.*", 5, 3, 2), MockRuleResult("gitlab", "bugfix/.*", 10, 0, 10) ]) print(f"Cycle {result.cycle_id}:") for rule_result in result.rule_results: print(f" {rule_result.provider} / {rule_result.rule_pattern}") print(f" Processed: {rule_result.processed}") print(f" Created: {rule_result.created}") print(f" Skipped: {rule_result.skipped_existing}") ```