From 02cd0429c03ce6b6099f04680423e153213a2f5e Mon Sep 17 00:00:00 2001 From: Dave Arnold Date: Wed, 30 Apr 2025 20:49:08 -0700 Subject: [PATCH] Refactor template automation models and add GitHub client implementation - Updated models in `models.py` to enhance documentation and structure. - Introduced `TemplateConfig` class to encapsulate PR and workflow configurations. - Modified `TemplateManager` in `template_manager.py` to utilize the new `TemplateConfig` model. - Implemented `GitHubClient` class in `github_client.py` for comprehensive GitHub API interactions, including repository creation, file management, and workflow triggering. - Improved error handling and logging throughout the GitHub client methods. --- docs/build_docs.sh | 1 + docs/source/modules/github_client.rst | 2 +- docs/source/modules/lambda_handler.rst | 2 +- requirements.txt | 7 + template_automation/app.py | 608 ++++++++---------------- template_automation/github_client.py | 452 ++++++++++++++++++ template_automation/models.py | 168 ++++++- template_automation/template_manager.py | 71 +-- 8 files changed, 836 insertions(+), 475 deletions(-) create mode 100644 template_automation/github_client.py diff --git a/docs/build_docs.sh b/docs/build_docs.sh index 5dfddb5..b855b42 100755 --- a/docs/build_docs.sh +++ b/docs/build_docs.sh @@ -19,6 +19,7 @@ mkdir -p docs/source/_static mkdir -p docs/build # Generate documentation +export SPHINX_BUILD=1 cd docs sphinx-build -b html source build diff --git a/docs/source/modules/github_client.rst b/docs/source/modules/github_client.rst index adf3168..3b5eb74 100644 --- a/docs/source/modules/github_client.rst +++ b/docs/source/modules/github_client.rst @@ -1,7 +1,7 @@ GitHub Client ============= -.. automodule:: template_automation.app +.. automodule:: template_automation.github_client :members: :undoc-members: :show-inheritance: diff --git a/docs/source/modules/lambda_handler.rst b/docs/source/modules/lambda_handler.rst index dcf10ad..9e1f993 100644 --- a/docs/source/modules/lambda_handler.rst +++ b/docs/source/modules/lambda_handler.rst @@ -2,6 +2,6 @@ Lambda Handler ============= .. automodule:: template_automation.app - :members: lambda_handler, operate_github, generate_repository_name, get_parameter + :members: lambda_handler, get_github_token :undoc-members: :show-inheritance: diff --git a/requirements.txt b/requirements.txt index 519c23c..701d88e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,11 @@ +# Core dependencies PyGithub>=2.1.1 pydantic~=2.6 boto3>=1.38.6 requests>=2.32.3 +jinja2>=3.1.0 +typing_extensions>=4.4.0 +pynacl>=1.5.0 # Required by PyGithub for cryptography +cryptography>=44.0.0 # Required by PyGithub for auth +pyjwt[crypto]>=2.10.0 # Required by PyGithub for JWT support +deprecated>=1.2.18 # Required by PyGithub for decorators diff --git a/template_automation/app.py b/template_automation/app.py index ed5bcf4..fa62b1c 100644 --- a/template_automation/app.py +++ b/template_automation/app.py @@ -1,432 +1,218 @@ -#################################################################################### -# This Lambda function creates new GitHub repositories from a template repository. -# It takes JSON input and writes it to a configurable config file in the new repo. -# Key features: -# - Template agnostic: Can be used with any type of template repository -# - Team-based admin access: Set owning team with full admin access -# - Configurable settings via Parameter Store (with prefix) or environment variables: -# - PARAM_STORE_PREFIX: Prefix for SSM parameters (default: /template-automation) -# - TEMPLATE_CONFIG_FILE: Name of config file to write (default: config.json) -# - TEMPLATE_TOPICS: Comma-separated list of topics to add (default: infrastructure) -# - TEMPLATE_REPO_NAME: Source template repository name (required) -# - TEMPLATE_SOURCE_VERSION: Version/tag/SHA to use from template (optional) -# - REPO_NAME_PREFIX: Prefix for generated repository names (optional) -# - GITHUB_API: GitHub API URL (required) -# - GITHUB_ORG_NAME: GitHub organization name (required) -# - GITHUB_COMMIT_AUTHOR_NAME: Name for commits (default: Template Automation) -# - GITHUB_COMMIT_AUTHOR_EMAIL: Email for commits (default: automation@example.com) -# - SECRET_NAME: AWS Secrets Manager secret containing GitHub token (required) -# -# Repository naming: -# - If REPO_NAME_PREFIX is set: Creates repos named {prefix}-{random-8-chars} -# - If not set: Uses the provided project name directly -# -# Implementation uses pure Python with requests library (no Git CLI dependency). -#################################################################################### +"""AWS Lambda function for creating GitHub repositories from templates. + +This module provides a Lambda function handler that automates the creation of new GitHub repositories +from a template repository. It validates input, writes configuration files, and creates pull requests +to set up the new repository. + +Features: + - Template agnostic: Works with any type of template repository + - Team-based access control: Automated team permission setup + - Configuration management: Writes template configuration via pull request + - Workflow automation: Optional initialization workflow trigger + - Pure Python implementation: No Git CLI dependency + - AWS integration: Uses Secrets Manager for token storage + +Configuration: + The Lambda function is configured through environment variables: + + Required: + GITHUB_API: Base URL for GitHub API server + GITHUB_ORG_NAME: Name of GitHub organization to manage + TEMPLATE_REPO_NAME: Name of the template repository to use + GITHUB_TOKEN_SECRET_NAME: Name of AWS Secrets Manager secret containing GitHub token + + Optional: + TEMPLATE_CONFIG_FILE: Name of config file to write (default: config.json) + TEMPLATE_TOPICS: Comma-separated topics to add (default: infrastructure) + PARAM_STORE_PREFIX: Prefix for SSM parameters (default: /template-automation) + GITHUB_COMMIT_AUTHOR_NAME: Name for commits (default: Template Automation) + GITHUB_COMMIT_AUTHOR_EMAIL: Email for commits (default: automation@example.com) + TEMPLATE_SOURCE_VERSION: Version/tag/SHA to use from template + +See Also: + - GitHubClient: Handles all GitHub API interactions + - TemplateManager: Manages template configuration and rendering + - models.TemplateInput: Validates Lambda function input +""" import os -import stat -import shutil import logging -import base64 import time -from datetime import datetime import traceback -import uuid -from github import Github, GithubException -from github.GithubObject import NotSet -from github.InputGitTreeElement import InputGitTreeElement +import boto3 from botocore.exceptions import ClientError -from .models import TemplateInput, GitHubConfig, WorkflowConfig +from typing import Optional +from .models import TemplateInput, GitHubConfig from .template_manager import TemplateManager +from .github_client import GitHubClient # Initialize the logger logger = logging.getLogger() logger.setLevel("INFO") # Set to "ERROR" to reduce logging messages. -# Get environment variables -GITHUB_TOKEN_SECRET_NAME = os.environ.get("GITHUB_TOKEN_SECRET_NAME") +# Required environment variables +REQUIRED_ENV_VARS = [ + "GITHUB_API", + "GITHUB_ORG_NAME", + "TEMPLATE_REPO_NAME", + "GITHUB_TOKEN_SECRET_NAME" +] + +# Check if we're being imported for documentation +IN_SPHINX_BUILD = os.environ.get('SPHINX_BUILD') == '1' + +# Skip validation during documentation build +if not IN_SPHINX_BUILD: + # Validate required environment variables + missing_vars = [var for var in REQUIRED_ENV_VARS if not os.environ.get(var)] + if missing_vars: + raise ValueError(f"Missing required environment variables: {', '.join(missing_vars)}") + +# Get environment variables with defaults for documentation +GITHUB_TOKEN_SECRET_NAME = os.environ.get("GITHUB_TOKEN_SECRET_NAME", "docs-placeholder") DEFAULT_CONFIG_FILE = os.environ.get("TEMPLATE_CONFIG_FILE", "config.json") DEFAULT_TOPICS = os.environ.get("TEMPLATE_TOPICS", "infrastructure").split(",") PARAM_STORE_PREFIX = os.environ.get("PARAM_STORE_PREFIX", "/template-automation") - -class GitHubClient: - """An object-oriented interface for GitHub repository management and automation. - - This class provides a high-level interface to GitHub's API using PyGithub, - enabling repository creation, content management, branch operations, and more. - It's specifically designed for template-based repository automation. - - Attributes: - github (Github): The PyGithub client instance. - org (Organization): The GitHub organization being operated on. - commit_author_name (str): The name to use for commit authorship. - commit_author_email (str): The email to use for commit authorship. - source_version (str, optional): The version, tag, or SHA of the template. - template_repo_name (str, optional): The name of the template repository. - config_file_name (str): The name of the configuration file to write. +TEMPLATE_SOURCE_VERSION: Optional[str] = os.environ.get("TEMPLATE_SOURCE_VERSION") + +# Keep imports and logging setup from here +# The GitHubClient class has been moved to github_client.py + +def lambda_handler(event: dict, context) -> dict: + """Process requests to create new repositories from templates. + + The handler executes the following steps: + 1. Validates the input event using TemplateInput model + 2. Retrieves GitHub token from AWS Secrets Manager + 3. Initializes GitHub client and template manager + 4. Creates new repository with team permissions + 5. Creates a feature branch for configuration + 6. Writes template configuration file + 7. Creates a pull request with the changes + 8. Optionally triggers initialization workflow + + Args: + event: AWS Lambda event containing: + project_name (str): Name for the new repository + template_settings (dict): Configuration values for the template + trigger_init_workflow (bool): Whether to trigger initialization + owning_team (str, optional): GitHub team to grant admin access + context: AWS Lambda context object (unused) + + Returns: + dict: Creation results containing: + repository_url (str): URL of the created repository + pull_request_url (str, optional): URL of the config pull request + + Raises: + ValueError: If input validation fails + ClientError: On AWS Secrets Manager errors + GithubException: On GitHub API errors + + Example: + >>> event = { + ... "project_name": "my-new-service", + ... "template_settings": { + ... "environment": "production", + ... "region": "us-west-2" + ... }, + ... "trigger_init_workflow": True, + ... "owning_team": "platform-team" + ... } + >>> result = lambda_handler(event, None) + >>> print(result["repository_url"]) + 'https://github.com/myorg/my-new-service' """ - - def __init__(self, config: GitHubConfig): - """Initialize the GitHub client with configuration. - - Args: - config (GitHubConfig): A Pydantic model containing validated GitHub configuration. - This includes the base URL, token, organization name, and other settings. - """ - self.github = Github( - base_url=config.api_base_url, - login_or_token=config.token + try: + logger.info(f"Processing template request: {event}") + + # Parse and validate input + template_input = TemplateInput(**event) + logger.info(f"Validated input for project: {template_input.project_name}") + + # Get GitHub configuration from environment/parameter store + github_config = GitHubConfig( + api_base_url=os.environ["GITHUB_API"], + org_name=os.environ["GITHUB_ORG_NAME"], + commit_author_name=os.environ.get("GITHUB_COMMIT_AUTHOR_NAME", "Template Automation"), + commit_author_email=os.environ.get("GITHUB_COMMIT_AUTHOR_EMAIL", "automation@example.com"), + token=get_github_token(), + template_repo_name=os.environ["TEMPLATE_REPO_NAME"], + config_file_name=DEFAULT_CONFIG_FILE, + source_version=TEMPLATE_SOURCE_VERSION ) - self.org = self.github.get_organization(config.org_name) - self.commit_author_name = config.commit_author_name - self.commit_author_email = config.commit_author_email - self.source_version = config.source_version - self.template_repo_name = config.template_repo_name - self.config_file_name = config.config_file_name - - def get_repository(self, repo_name: str, create: bool = False, owning_team: str = None): - """Get or create a GitHub repository with optional team permissions. - - This method attempts to retrieve a repository by name. If it doesn't exist and - create=True, it creates a new repository. It also handles team permissions - if an owning team is specified. - - Args: - repo_name (str): The name of the repository to retrieve or create. - create (bool, optional): Whether to create the repository if it doesn't exist. - owning_team (str, optional): The name of the GitHub team to grant admin access. - - Returns: - github.Repository.Repository: The repository object. - - Raises: - GithubException: If repository operations fail. - """ - try: - try: - repo = self.org.get_repo(repo_name) - logger.info(f"Found existing repository: {repo_name}") - if owning_team: - self.set_team_permission(repo_name, owning_team, "admin") - return repo - except GithubException as e: - if e.status == 404 and create: - logger.info(f"Creating repository {repo_name}") - repo = self.org.create_repo( - name=repo_name, - private=True, - auto_init=True, - allow_squash_merge=True, - allow_merge_commit=True, - allow_rebase_merge=True, - delete_branch_on_merge=True - ) - - # Wait for repository initialization - max_retries = 100 - retry_delay = 1 - for _ in range(max_retries): - try: - repo.get_branch("main") - break - except GithubException: - time.sleep(retry_delay) - else: - raise Exception(f"Repository {repo_name} initialization timed out") - - if owning_team: - self.set_team_permission(repo_name, owning_team, "admin") - return repo - raise - except GithubException as e: - error_message = f"GitHub API error: {str(e)}" - logger.error(error_message) - raise - - def get_default_branch(self, repo_name: str) -> str: - """Get the default branch name of a repository. - - Args: - repo_name (str): Name of the repository. - - Returns: - str: Default branch name (usually 'main' or 'master'). - """ - repo = self.org.get_repo(repo_name) - return repo.default_branch - - def create_branch(self, repo_name: str, branch_name: str, from_ref: str = "main") -> None: - """Create a new branch in the repository. - - Args: - repo_name (str): Name of the repository. - branch_name (str): Name of the branch to create. - from_ref (str, optional): Reference to create branch from. Defaults to "main". - - Raises: - GithubException: If branch creation fails. - """ - repo = self.org.get_repo(repo_name) - source = repo.get_branch(from_ref) - try: - repo.create_git_ref( - ref=f"refs/heads/{branch_name}", - sha=source.commit.sha - ) - logger.info(f"Created branch {branch_name} in {repo_name}") - except GithubException as e: - error_message = f"Failed to create branch {branch_name}: {str(e)}" - logger.error(error_message) - raise - - def create_commit(self, repo_name: str, branch: str, commit_message: str, changes: list) -> None: - """Create a commit with the specified changes. - - Args: - repo_name (str): Name of the repository. - branch (str): Branch name to commit to. - commit_message (str): Commit message. - changes (list): List of dictionaries with 'path' and 'content' keys. - - Raises: - GithubException: If commit creation fails. - """ - repo = self.org.get_repo(repo_name) + # Initialize clients + github = GitHubClient(github_config) + template_mgr = TemplateManager(github_config) - try: - # Get the branch reference - ref = repo.get_git_ref(f"heads/{branch}") - branch_sha = ref.object.sha - base_tree = repo.get_git_tree(branch_sha) - - # Create tree elements - tree_elements = [] - for change in changes: - element = InputGitTreeElement( - path=change['path'], - mode='100644', - type='blob', - content=change['content'] - ) - tree_elements.append(element) - - # Create tree - tree = repo.create_git_tree(tree_elements, base_tree) - - # Create commit - parent = repo.get_git_commit(branch_sha) - commit = repo.create_git_commit( - message=commit_message, - tree=tree, - parents=[parent], - author={"name": self.commit_author_name, "email": self.commit_author_email}, - committer={"name": self.commit_author_name, "email": self.commit_author_email} - ) - - # Update branch reference - ref.edit(commit.sha, force=True) - - logger.info(f"Created commit in {repo_name}/{branch}: {commit_message}") - except GithubException as e: - error_message = f"Failed to create commit: {str(e)}" - logger.error(error_message) - raise - - def clone_repository_contents(self, source_repo: str, target_repo: str, - source_ref: str = None, target_branch: str = "main") -> None: - """Clone contents from one repository to another using PyGithub. - - Args: - source_repo (str): Name of the source repository. - target_repo (str): Name of the target repository. - source_ref (str, optional): Source ref (branch/tag/SHA). Defaults to default branch. - target_branch (str, optional): Target branch name. Defaults to "main". - - Raises: - GithubException: If repository operations fail. - """ - try: - source = self.org.get_repo(source_repo) - target = self.org.get_repo(target_repo) - - # If no source_ref specified, use the default branch - if not source_ref: - source_ref = source.default_branch - - # Get the tree from source repository - if source_ref.startswith('refs/'): - ref = source.get_git_ref(source_ref.replace('refs/', '')) - tree_sha = ref.object.sha - else: - # Try as a branch first - try: - branch = source.get_branch(source_ref) - tree_sha = branch.commit.sha - except GithubException: - # If not a branch, try as a commit SHA - tree_sha = source_ref - - # Get the full tree - tree = source.get_git_tree(tree_sha, recursive=True) - - # Download and create all blobs - elements = [] - for entry in tree.tree: - if entry.type == 'blob': - blob = source.get_git_blob(entry.sha) - content = base64.b64decode(blob.content).decode('utf-8') - elements.append({ - 'path': entry.path, - 'content': content - }) - - # Create commit with all files - if elements: - self.create_commit( - target_repo, - target_branch, - f"Clone contents from {source_repo}", - elements - ) - - logger.info(f"Successfully cloned contents from {source_repo} to {target_repo}") - except GithubException as e: - error_message = f"Failed to clone repository contents: {str(e)}" - logger.error(error_message) - raise - - def commit_repository_contents(self, repo_name: str, branch: str, contents: dict) -> None: - """Commit multiple file contents to a repository. - - Args: - repo_name (str): Name of the repository. - branch (str): Branch to commit to. - contents (dict): Dictionary mapping file paths to their content. - - Raises: - GithubException: If commit operations fail. - """ - try: - # Format changes for create_commit method - changes = [ - {'path': path, 'content': content} - for path, content in contents.items() - ] - - self.create_commit( + # Create repository from template + repo_name = template_input.project_name + repo = github.get_repository(repo_name, create=True, owning_team=template_input.owning_team) + + # Create feature branch for template configuration + feature_branch = f"template-config-{int(time.time())}" + github.create_branch(repo_name, feature_branch) + + # Write template configuration + github.write_file( + repo=repo, + path=DEFAULT_CONFIG_FILE, + content=template_input.template_settings.json(), + branch=feature_branch, + commit_message=f"Initialize {DEFAULT_CONFIG_FILE} from template" + ) + + # Set repository topics + github.update_repository_topics(repo_name, DEFAULT_TOPICS) + + # Create pull request with template configuration + config = template_mgr.get_pr_config() + pr_title = config.title_template.format(repo_name=repo_name) + pr_body = config.body_template.format( + repo_name=repo_name, + template_repo=github_config.template_repo_name + ) + + pr = github.create_pull_request( + repo_name=repo_name, + title=pr_title, + body=pr_body, + head_branch=feature_branch, + base_branch=github.get_default_branch(repo_name) + ) + + # Optionally trigger initialization workflow + if template_input.trigger_init_workflow: + github.trigger_workflow( repo_name=repo_name, - branch=branch, - commit_message="Update repository contents", - changes=changes + workflow_id="initialize.yml", + ref=feature_branch ) - - logger.info(f"Successfully committed contents to {repo_name}/{branch}") - except GithubException as e: - error_message = f"Failed to commit repository contents: {str(e)}" - logger.error(error_message) - raise - - def update_repository_topics(self, repo_name: str, topics: list) -> None: - """Update the topics of a repository. - - Args: - repo_name (str): Name of the repository. - topics (list): List of topics to set. - - Raises: - GithubException: If the operation fails. - """ - try: - repo = self.org.get_repo(repo_name) - repo.replace_topics(topics) - logger.info(f"Updated topics for {repo_name}: {topics}") - except GithubException as e: - error_message = f"Failed to update repository topics: {str(e)}" - logger.error(error_message) - raise - - def set_team_permission(self, repo_name: str, team_name: str, permission: str) -> None: - """Set a team's permission on a repository. - - Args: - repo_name (str): Name of the repository. - team_name (str): Name of the team. - permission (str): Permission level ('pull', 'push', 'admin', 'maintain', 'triage'). - - Raises: - GithubException: If the operation fails. - """ - try: - repo = self.org.get_repo(repo_name) - team = self.org.get_team_by_slug(team_name) - team.update_team_repository(repo, permission) - logger.info(f"Set {team_name} permission on {repo_name} to {permission}") - except GithubException as e: - error_message = f"Failed to set team permission: {str(e)}" - logger.error(error_message) - raise - - def create_pull_request(self, repo_name: str, title: str, body: str, - head_branch: str, base_branch: str = "main") -> None: - """Create a pull request in a repository. - - Args: - repo_name (str): Name of the repository. - title (str): Title of the pull request. - body (str): Description/body of the pull request. - head_branch (str): Branch containing the changes. - base_branch (str, optional): Branch to merge into. Defaults to "main". - - Returns: - github.PullRequest.PullRequest: The created pull request object. - - Raises: - GithubException: If pull request creation fails. - """ - try: - repo = self.org.get_repo(repo_name) - pr = repo.create_pull( - title=title, - body=body, - head=head_branch, - base=base_branch, - maintainer_can_modify=True - ) - logger.info(f"Created PR #{pr.number} in {repo_name}: {title}") - return pr - except GithubException as e: - error_message = f"Failed to create pull request: {str(e)}" - logger.error(error_message) - raise - - def trigger_workflow(self, repo_name: str, workflow_id: str, ref: str, - inputs: dict = None) -> None: - """Trigger a GitHub Actions workflow. - - Args: - repo_name (str): Name of the repository. - workflow_id (str): ID or filename of the workflow. - ref (str): Git reference to run the workflow on. - inputs (dict, optional): Input parameters for the workflow. - - Raises: - GithubException: If workflow dispatch fails. - """ - try: - repo = self.org.get_repo(repo_name) - workflow = repo.get_workflow(workflow_id) - - # Convert inputs to GitHub's expected format - workflow_inputs = inputs if inputs is not None else {} - - workflow.create_dispatch( - ref=ref, - inputs=workflow_inputs - ) - logger.info(f"Triggered workflow {workflow_id} in {repo_name} on {ref}") - except GithubException as e: - error_message = f"Failed to trigger workflow: {str(e)}" - logger.error(error_message) - raise + + return { + "repository_url": repo.html_url, + "pull_request_url": pr.html_url if pr else None + } + + except Exception as e: + logger.error(f"Failed to process template request: {str(e)}") + logger.error(traceback.format_exc()) + raise + +def get_github_token() -> str: + """Get GitHub token from AWS Secrets Manager. + + Returns: + str: GitHub API token + + Raises: + ClientError: If secret retrieval fails + """ + try: + client = boto3.client('secretsmanager') + response = client.get_secret_value(SecretId=GITHUB_TOKEN_SECRET_NAME) + return response['SecretString'] + except ClientError as e: + logger.error(f"Failed to get GitHub token: {str(e)}") + raise diff --git a/template_automation/github_client.py b/template_automation/github_client.py new file mode 100644 index 0000000..da7a163 --- /dev/null +++ b/template_automation/github_client.py @@ -0,0 +1,452 @@ +"""GitHub client module for template automation. + +This module provides the GitHubClient class which handles all interactions with the GitHub API +for template repository automation. +""" + +import base64 +import logging +import time +from typing import List, Optional, Dict, Any, Union + +from github import Github, GithubException +from github.Repository import Repository +from github.ContentFile import ContentFile +from github.Organization import Organization +from github.Team import Team +from github.PullRequest import PullRequest +from github.Workflow import Workflow + +logger = logging.getLogger(__name__) + +class GitHubClient: + """A client for interacting with GitHub's API in the context of template automation. + + This class provides methods for template repository operations including: + - Creating repositories from templates + - Managing repository contents + - Setting up team access + - Configuring repository settings + + Attributes: + api_base_url (str): Base URL for the GitHub API + token (str): GitHub authentication token + org_name (str): GitHub organization name + commit_author_name (str): Name to use for automated commits + commit_author_email (str): Email to use for automated commits + client (Github): PyGithub client instance + org (Organization): GitHub organization instance + + Example: + ```python + client = GitHubClient( + api_base_url="https://api.github.com", + token="ghp_...", + org_name="my-org", + commit_author_name="Template Bot", + commit_author_email="bot@example.com" + ) + + repo = client.create_repository_from_template( + template_repo_name="template-service", + new_repo_name="new-service", + private=True + ) + ``` + """ + + def __init__( + self, + api_base_url: str, + token: str, + org_name: str, + commit_author_name: str = "Template Automation", + commit_author_email: str = "automation@example.com" + ): + """Initialize a new GitHub client. + + Args: + api_base_url: Base URL for the GitHub API + token: GitHub authentication token + org_name: GitHub organization name + commit_author_name: Name to use for automated commits + commit_author_email: Email to use for automated commits + + Raises: + GithubException: If authentication fails or org doesn't exist + """ + self.api_base_url = api_base_url + self.token = token + self.org_name = org_name + self.commit_author_name = commit_author_name + self.commit_author_email = commit_author_email + + # Initialize client and get org + self.client = Github(base_url=api_base_url, login_or_token=token) + self.org = self.client.get_organization(org_name) + logger.info(f"Initialized GitHub client for org: {org_name}") + + def create_repository_from_template( + self, + template_repo_name: str, + new_repo_name: str, + private: bool = True, + description: Optional[str] = None, + topics: Optional[List[str]] = None + ) -> Repository: + """Create a new repository from a template. + + Args: + template_repo_name: Name of the template repository + new_repo_name: Name for the new repository + private: Whether the new repository should be private + description: Description for the new repository + topics: List of topics to add to the repository + + Returns: + The newly created repository + + Raises: + GithubException: If template doesn't exist or repo creation fails + """ + template_repo = self.org.get_repo(template_repo_name) + + # Create repository from template + new_repo = self.org.create_repository_from_template( + name=new_repo_name, + template_repository=template_repo, + private=private, + description=description or f"Repository created from template: {template_repo_name}" + ) + + # Add topics if provided + if topics: + new_repo.replace_topics(topics) + + logger.info(f"Created new repository: {new_repo_name} from template: {template_repo_name}") + return new_repo + + def set_team_access(self, repo: Repository, team_slug: str, permission: str = "admin") -> None: + """Give a team access to a repository. + + Args: + repo: The repository to grant access to + team_slug: The team's slug identifier + permission: The permission level to grant (pull, push, admin) + + Raises: + GithubException: If team doesn't exist or permission grant fails + """ + try: + team = self.org.get_team_by_slug(team_slug) + team.add_to_repos(repo) + team.set_repo_permission(repo, permission) + logger.info(f"Granted {permission} access to team {team_slug} for repo {repo.name}") + except GithubException as e: + logger.error(f"Failed to set team access: {e}") + raise + + def write_file( + self, + repo: Repository, + path: str, + content: str, + branch: str = "main", + commit_message: Optional[str] = None + ) -> ContentFile: + """Write or update a file in a repository. + + Args: + repo: The repository to write to + path: Path where to create/update the file + content: Content to write to the file + branch: Branch to commit to + commit_message: Commit message to use + + Returns: + The created/updated file content + + Raises: + GithubException: If file operation fails + """ + try: + # Convert content to base64 + content_bytes = content.encode("utf-8") + content_base64 = base64.b64encode(content_bytes).decode("utf-8") + + # Try to get existing file + try: + file = repo.get_contents(path, ref=branch) + # Update existing file + result = repo.update_file( + path=path, + message=commit_message or f"Update {path}", + content=content_base64, + sha=file.sha, + branch=branch, + committer={ + "name": self.commit_author_name, + "email": self.commit_author_email + } + ) + logger.info(f"Updated file {path} in repo {repo.name}") + return result["content"] + except GithubException as e: + if e.status != 404: # Only handle "not found" errors + raise + + # Create new file + result = repo.create_file( + path=path, + message=commit_message or f"Create {path}", + content=content_base64, + branch=branch, + committer={ + "name": self.commit_author_name, + "email": self.commit_author_email + } + ) + logger.info(f"Created new file {path} in repo {repo.name}") + return result["content"] + + except GithubException as e: + logger.error(f"Failed to write file {path}: {e}") + raise + + def read_file( + self, + repo: Repository, + path: str, + ref: str = "main" + ) -> str: + """Read a file from a repository. + + Args: + repo: The repository to read from + path: Path to the file to read + ref: Git reference (branch, tag, commit) to read from + + Returns: + The file contents as a string + + Raises: + GithubException: If file doesn't exist or read fails + """ + try: + file = repo.get_contents(path, ref=ref) + content = base64.b64decode(file.content).decode("utf-8") + return content + except GithubException as e: + logger.error(f"Failed to read file {path}: {e}") + raise + + def get_repository( + self, + repo_name: str, + create: bool = False, + owning_team: Optional[str] = None + ) -> Repository: + """Get or create a GitHub repository with optional team permissions. + + Args: + repo_name: The name of the repository to retrieve or create + create: Whether to create the repository if it doesn't exist + owning_team: The name of the GitHub team to grant admin access + + Returns: + The repository object + + Raises: + GithubException: If repository operations fail + """ + try: + try: + repo = self.org.get_repo(repo_name) + logger.info(f"Found existing repository: {repo_name}") + if owning_team: + self.set_team_permission(repo_name, owning_team, "admin") + return repo + except GithubException as e: + if e.status == 404 and create: + logger.info(f"Creating repository {repo_name}") + repo = self.org.create_repo( + name=repo_name, + private=True, + auto_init=True, + allow_squash_merge=True, + allow_merge_commit=True, + allow_rebase_merge=True, + delete_branch_on_merge=True + ) + + # Wait for repository initialization + max_retries = 100 + retry_delay = 1 + for _ in range(max_retries): + try: + repo.get_branch("main") + break + except GithubException: + time.sleep(retry_delay) + else: + raise Exception(f"Repository {repo_name} initialization timed out") + + if owning_team: + self.set_team_permission(repo_name, owning_team, "admin") + return repo + raise + except GithubException as e: + error_message = f"GitHub API error: {str(e)}" + logger.error(error_message) + raise + + def get_default_branch(self, repo_name: str) -> str: + """Get the default branch name of a repository. + + Args: + repo_name: Name of the repository + + Returns: + Default branch name (usually 'main' or 'master') + """ + repo = self.org.get_repo(repo_name) + return repo.default_branch + + def create_branch(self, repo_name: str, branch_name: str, from_ref: str = "main") -> None: + """Create a new branch in the repository. + + Args: + repo_name: Name of the repository + branch_name: Name of the branch to create + from_ref: Reference to create branch from + + Raises: + GithubException: If branch creation fails + """ + repo = self.org.get_repo(repo_name) + source = repo.get_branch(from_ref) + + try: + repo.create_git_ref( + ref=f"refs/heads/{branch_name}", + sha=source.commit.sha + ) + logger.info(f"Created branch {branch_name} in {repo_name}") + except GithubException as e: + error_message = f"Failed to create branch {branch_name}: {str(e)}" + logger.error(error_message) + raise + + def create_pull_request( + self, + repo_name: str, + title: str, + body: str, + head_branch: str, + base_branch: str = "main" + ) -> Any: + """Create a pull request in a repository. + + Args: + repo_name: Name of the repository + title: Title of the pull request + body: Description/body of the pull request + head_branch: Branch containing the changes + base_branch: Branch to merge into + + Returns: + The created pull request object + + Raises: + GithubException: If pull request creation fails + """ + try: + repo = self.org.get_repo(repo_name) + pr = repo.create_pull( + title=title, + body=body, + head=head_branch, + base=base_branch, + maintainer_can_modify=True + ) + logger.info(f"Created PR #{pr.number} in {repo_name}: {title}") + return pr + except GithubException as e: + error_message = f"Failed to create pull request: {str(e)}" + logger.error(error_message) + raise + + def trigger_workflow( + self, + repo_name: str, + workflow_id: str, + ref: str, + inputs: Optional[Dict[str, Any]] = None + ) -> None: + """Trigger a GitHub Actions workflow. + + Args: + repo_name: Name of the repository + workflow_id: ID or filename of the workflow + ref: Git reference to run the workflow on + inputs: Input parameters for the workflow + + Raises: + GithubException: If workflow dispatch fails + """ + try: + repo = self.org.get_repo(repo_name) + workflow = repo.get_workflow(workflow_id) + + # Convert inputs to GitHub's expected format + workflow_inputs = inputs if inputs is not None else {} + + workflow.create_dispatch( + ref=ref, + inputs=workflow_inputs + ) + logger.info(f"Triggered workflow {workflow_id} in {repo_name} on {ref}") + except GithubException as e: + error_message = f"Failed to trigger workflow: {str(e)}" + logger.error(error_message) + raise + + def set_team_permission(self, repo_name: str, team_name: str, permission: str) -> None: + """Set a team's permission on a repository. + + Args: + repo_name: Name of the repository + team_name: Name of the team + permission: Permission level ('pull', 'push', 'admin', 'maintain', 'triage') + + Raises: + GithubException: If the operation fails + """ + try: + repo = self.org.get_repo(repo_name) + team = self.org.get_team_by_slug(team_name) + team.update_team_repository(repo, permission) + logger.info(f"Set {team_name} permission on {repo_name} to {permission}") + except GithubException as e: + error_message = f"Failed to set team permission: {str(e)}" + logger.error(error_message) + raise + + def update_repository_topics(self, repo_name: str, topics: List[str]) -> None: + """Update the topics of a repository. + + Args: + repo_name: Name of the repository + topics: List of topics to set + + Raises: + GithubException: If the operation fails + """ + try: + repo = self.org.get_repo(repo_name) + repo.replace_topics(topics) + logger.info(f"Updated topics for {repo_name}: {topics}") + except GithubException as e: + error_message = f"Failed to update repository topics: {str(e)}" + logger.error(error_message) + raise diff --git a/template_automation/models.py b/template_automation/models.py index 9208229..922ebbd 100644 --- a/template_automation/models.py +++ b/template_automation/models.py @@ -1,7 +1,7 @@ -"""Models for template automation using Pydantic.""" +"""Models for template automation.""" +from typing import List, Dict, Any, Optional from pydantic import BaseModel, Field -from typing import Dict, List, Optional, Any class GitHubConfig(BaseModel): """Encapsulates configuration settings for interacting with the GitHub API. @@ -28,11 +28,33 @@ class GitHubConfig(BaseModel): class WorkflowConfig(BaseModel): """Defines the configuration for a GitHub Actions workflow. + This class represents a single GitHub Actions workflow configuration, + including its template source and destination paths, along with any + variables needed for template rendering. + Attributes: - name (str): The name of the workflow. - template_path (str): The path to the workflow template file. - output_path (str): The path where the rendered workflow file will be saved. - variables (Dict[str, Any]): A dictionary of variables to substitute in the template. + name (str): Descriptive name of the workflow, used for logging and + identification purposes. + template_path (str): Path to the Jinja2 template file containing the + workflow definition. This path should be relative to the template + root directory. + output_path (str): Destination path where the rendered workflow file + will be saved in the new repository. This path should be relative + to the repository root. + variables (Dict[str, Any]): Dictionary of variables to use when + rendering the workflow template. These values will be passed to + the Jinja2 template engine. + + Example: + >>> workflow = WorkflowConfig( + ... name="CI/CD Pipeline", + ... template_path="workflows/ci.yml.j2", + ... output_path=".github/workflows/ci.yml", + ... variables={ + ... "python_version": "3.9", + ... "test_commands": ["pytest", "flake8"] + ... } + ... ) """ name: str template_path: str @@ -42,14 +64,29 @@ class WorkflowConfig(BaseModel): class PRConfig(BaseModel): """Specifies the configuration for creating pull requests. + This class defines the structure and default values for pull request creation, + including templates for title and body, branch configuration, and PR metadata + like labels and reviewers. + Attributes: - title_template (str): The template for the pull request title. - body_template (str): The template for the pull request body. - base_branch (str): The base branch for the pull request. - branch_prefix (str): The prefix for the branch name. - labels (List[str]): A list of labels to apply to the pull request. - reviewers (List[str]): A list of reviewers to request for the pull request. - assignees (List[str]): A list of assignees for the pull request. + title_template (str): Jinja2 template for the pull request title. Variables + available include: repo_name, template_repo. + body_template (str): Jinja2 template for the pull request body. Variables + available include: repo_name, template_repo, workflow_files. + base_branch (str): The target branch for the pull request. Defaults to "main". + branch_prefix (str): Prefix for the feature branch name. The final branch name + will be {prefix}-{repo_name}. + labels (List[str]): Labels to automatically apply to the pull request. + Defaults to ["automated"]. + reviewers (List[str]): GitHub usernames of reviewers to assign. + assignees (List[str]): GitHub usernames of users to assign to the PR. + + Example: + >>> pr_config = PRConfig( + ... title_template="Initialize {{ repo_name }} from template", + ... labels=["infrastructure", "automated"], + ... reviewers=["alice", "bob"] + ... ) """ title_template: str = "Initialize {{ repo_name }} from template" body_template: str = """ @@ -76,13 +113,110 @@ class PRConfig(BaseModel): class TemplateInput(BaseModel): """Represents the input data required for template automation. + This class defines the structure of input data needed to create a new + repository from a template. It includes project metadata, template-specific + settings, and optional configurations for repository ownership and + initialization. + Attributes: - project_name (str): The name of the project to create. - template_settings (Dict[str, Any]): A dictionary of settings for the template. - trigger_init_workflow (bool): Whether to trigger the initialization workflow. - owning_team (Optional[str]): The name of the team that will own the repository. + project_name (str): Name of the project/repository to create. This will + be used as the repository name and in various template substitutions. + template_settings (Dict[str, Any]): Dictionary of template-specific + settings that will be written to the configuration file in the new + repository. The structure depends on the template being used. + trigger_init_workflow (bool): Whether to automatically trigger the + initialization workflow after repository creation. Defaults to False. + owning_team (Optional[str]): The GitHub team slug that should be granted + admin access to the new repository. If None, no team access will be + configured. + + Example: + >>> input_data = TemplateInput( + ... project_name="my-new-service", + ... template_settings={ + ... "environment": "production", + ... "region": "us-west-2" + ... }, + ... trigger_init_workflow=True, + ... owning_team="platform-team" + ... ) """ project_name: str template_settings: Dict[str, Any] trigger_init_workflow: bool = False owning_team: Optional[str] = None + +class TemplateConfig(BaseModel): + """Configuration for a template repository. + + This class defines the configuration structure for template repositories, + including pull request settings and workflow configurations. + + Attributes: + pr (PRConfig): Pull request configuration settings including title template, + body template, branch settings, labels, reviewers and assignees. + workflows (List[WorkflowConfig]): List of workflow configurations to apply + to the repository. Each workflow config specifies name, template path, + output path and variables. + + Example: + ```python + config = TemplateConfig( + pr=PRConfig( + title_template="Initialize {{ repo_name }}", + base_branch="main", + labels=["automated"] + ), + workflows=[ + WorkflowConfig( + name="CI", + template_path="workflows/ci.yml", + output_path=".github/workflows/ci.yml" + ) + ] + ) + ``` + """ + pr: PRConfig = Field( + default_factory=lambda: PRConfig( + title_template="Initialize {{ repo_name }} from template", + body_template=""" + Automated pull request for initializing {{ repo_name }} from template {{ template_repo }}. + + This PR was created by the Template Automation system. + {% if workflow_files %} + ## Added Workflows + {% for workflow in workflow_files %} + - {{ workflow }} + {% endfor %} + {% endif %} + """, + base_branch="main", + branch_prefix="init", + labels=["automated"], + reviewers=[], + assignees=[] + ) + ) + workflows: List[WorkflowConfig] = Field(default_factory=list) + + class Config: + """Pydantic model configuration. + + This inner class defines metadata for the TemplateConfig model, + including example configurations and schema information. + """ + json_schema_extra = { + "example": { + "pr": { + "title_template": "Initialize {{ repo_name }} from template", + "body_template": "Template PR body...", + "base_branch": "main", + "branch_prefix": "init", + "labels": ["automated"], + "reviewers": [], + "assignees": [] + }, + "workflows": [] + } + } diff --git a/template_automation/template_manager.py b/template_automation/template_manager.py index 3aae770..8951790 100644 --- a/template_automation/template_manager.py +++ b/template_automation/template_manager.py @@ -2,9 +2,10 @@ import os import json -from typing import Dict, Any, List +from typing import Dict, Any, List, Optional from jinja2 import Environment, FileSystemLoader, Template -from .models import WorkflowConfig, PRConfig +from pydantic import ValidationError +from .models import WorkflowConfig, PRConfig, TemplateConfig class TemplateManager: """Handles the management and rendering of templates for workflows and pull requests. @@ -15,10 +16,10 @@ class TemplateManager: Attributes: env (Environment): The Jinja2 environment for rendering templates. template_repo_name (str): The name of the template repository. - config (Dict): The loaded template configuration. + config (TemplateConfig): The loaded template configuration. """ - def __init__(self, template_root: str = None, template_repo_name: str = None): + def __init__(self, template_root: Optional[str] = None, template_repo_name: Optional[str] = None): """Initialize the TemplateManager with optional template root and repository name. Args: @@ -37,48 +38,28 @@ def __init__(self, template_root: str = None, template_repo_name: str = None): self.template_repo_name = template_repo_name self.config = self._load_template_config() - def _load_template_config(self) -> Dict: + def _load_template_config(self) -> TemplateConfig: """Load the template configuration from a .template-config.json file. Returns: - Dict: The loaded configuration, merged with default settings. + TemplateConfig: The loaded configuration with validation. + + Raises: + ValidationError: If the configuration is invalid. """ - # Default configuration - default_config = { - "pr": { - "title_template": "Initialize {{ repo_name }} from template", - "body_template": """ - Automated pull request for initializing {{ repo_name }} from template {{ template_repo }}. - - This PR was created by the Template Automation system. - {% if workflow_files %} - ## Added Workflows - {% for workflow in workflow_files %} - - {{ workflow }} - {% endfor %} - {% endif %} - """, - "base_branch": "main", - "branch_prefix": "init", - "labels": ["automated"], - "reviewers": [], - "assignees": [] - }, - "workflows": [] # List of workflow configurations to apply - } - - # Try to load template-specific configuration try: config_path = os.path.join(os.getcwd(), ".template-config.json") if os.path.exists(config_path): with open(config_path, "r") as f: template_config = json.load(f) - # Merge with defaults, keeping template-specific values - default_config.update(template_config) + return TemplateConfig(**template_config) + return TemplateConfig() # Use defaults if no config file exists + except ValidationError as e: + print(f"Warning: Template config validation failed: {str(e)}") + return TemplateConfig() # Use defaults on validation error except Exception as e: print(f"Warning: Could not load template config: {str(e)}") - - return default_config + return TemplateConfig() # Use defaults on any other error def render_workflow(self, workflow: WorkflowConfig) -> str: """Render a GitHub Actions workflow template. @@ -92,7 +73,7 @@ def render_workflow(self, workflow: WorkflowConfig) -> str: template = self.env.get_template(workflow.template_path) return template.render(**workflow.variables) - def render_pr_details(self, repo_name: str, workflow_files: List[str] = None) -> Dict[str, Any]: + def render_pr_details(self, repo_name: str, workflow_files: Optional[List[str]] = None) -> Dict[str, Any]: """Generate pull request details by rendering templates and configurations. Args: @@ -102,7 +83,7 @@ def render_pr_details(self, repo_name: str, workflow_files: List[str] = None) -> Returns: Dict[str, Any]: A dictionary containing the rendered pull request details. """ - pr_config = self.config["pr"] + pr_config = self.config.pr variables = { "repo_name": repo_name, "template_repo": self.template_repo_name, @@ -110,13 +91,13 @@ def render_pr_details(self, repo_name: str, workflow_files: List[str] = None) -> } return { - "title": self.env.from_string(pr_config["title_template"]).render(**variables), - "body": self.env.from_string(pr_config["body_template"]).render(**variables), - "base_branch": pr_config["base_branch"], - "branch_name": f"{pr_config['branch_prefix']}-{repo_name}", - "labels": pr_config["labels"], - "reviewers": pr_config["reviewers"], - "assignees": pr_config["assignees"] + "title": self.env.from_string(pr_config.title_template).render(**variables), + "body": self.env.from_string(pr_config.body_template).render(**variables), + "base_branch": pr_config.base_branch, + "branch_name": f"{pr_config.branch_prefix}-{repo_name}", + "labels": pr_config.labels, + "reviewers": pr_config.reviewers, + "assignees": pr_config.assignees } def get_workflow_configs(self) -> List[WorkflowConfig]: @@ -125,4 +106,4 @@ def get_workflow_configs(self) -> List[WorkflowConfig]: Returns: List[WorkflowConfig]: A list of workflow configurations. """ - return [WorkflowConfig(**w) for w in self.config.get("workflows", [])] + return self.config.workflows