Merge branch 'main' of https://github.com/issue-to-pr/issue-to-pr int…

…o nikita

.gitignore

@@ -8,10 +8,5 @@ __pycache__
 # Virtual Environment
 venv
 
-venv/
-venv*
-.venv
-/venv
-/venv/
 # Others
 .aider.tags.cache.v3

config.py

@@ -6,19 +6,25 @@
 from dotenv import load_dotenv
 load_dotenv()
 
+
+# Function to get environment variable with error handling
+def get_env_var(name: str) -> str:
+    value: str | None = os.getenv(key=name)
+    if value is None:
+        raise ValueError(f"Environment variable {name} not set.")
+    return value
+
+
 # GitHub Credentials from environment variables
-GITHUB_APP_ID = os.environ.get("APP_ID_GITHUB")
-# GITHUB_CLIENT_ID = os.getenv("GITHUB_CLIENT_ID")
-# GITHUB_CLIENT_SECRET = os.getenv("GITHUB_CLIENT_SECRET")
-# GITHUB_INSTALLATION_ID = os.getenv("GITHUB_INSTALLATION_ID")
-GITHUB_PRIVATE_KEY_ENCODED = os.environ.get("GITHUB_PRIVATE_KEY")
-GITHUB_PRIVATE_KEY = ''#base64.b64decode(GITHUB_PRIVATE_KEY_ENCODED)
-GITHUB_WEBHOOK_SECRET = os.environ.get("GITHUB_WEBHOOK_SECRET")
+GITHUB_APP_ID: str = get_env_var(name="GITHUB_APP_ID")
+GITHUB_PRIVATE_KEY_ENCODED: str = get_env_var(name="GITHUB_PRIVATE_KEY")
+GITHUB_PRIVATE_KEY: bytes = base64.b64decode(s=GITHUB_PRIVATE_KEY_ENCODED)
+GITHUB_WEBHOOK_SECRET: str = get_env_var(name="GITHUB_WEBHOOK_SECRET")
+
 
 # Supabase Credentials from environment variables
-SUPABASE_URL = os.environ.get("SUPABASE_URL")
-SUPABASE_SERVICE_ROLE_KEY = os.environ.get("SUPABASE_SERVICE_ROLE_KEY")
-SUPABASE_JWT_SECRET_KEY = os.environ.get("SUPABASE_JWT_SECRET_KEY")
+SUPABASE_URL: str = get_env_var(name="SUPABASE_URL")
+SUPABASE_SERVICE_ROLE_KEY: str = get_env_var(name="SUPABASE_SERVICE_ROLE_KEY")
 
 # General
 LABEL = "pragent"

main.py

@@ -1,40 +1,49 @@
-
-
+# Standard imports
 import json
 
+# Third-party imports
 from fastapi import FastAPI, HTTPException, Request
-import urllib.parse
-
-from config import GITHUB_APP_ID
+from mangum import Mangum
 
+# Local imports
+from config import GITHUB_APP_ID, GITHUB_PRIVATE_KEY, GITHUB_WEBHOOK_SECRET
+from services.github.github_manager import GitHubManager
 from services.github.webhook_handler import handle_webhook_event
 
-from mangum import Mangum
-app = FastAPI( title="hello world",
-    openapi_prefix="/prod")
-handler = Mangum(app)
+# Create FastAPI instance
+app = FastAPI()
+handler = Mangum(app=app)
 
+# Initialize GitHub manager
+github_manager = GitHubManager(app_id=GITHUB_APP_ID, private_key=GITHUB_PRIVATE_KEY)
 
-@app.post("/webhook")
-async def handle_webhook(request: Request):
+
+@app.post(path="/webhook")
+async def handle_webhook(request: Request) -> dict[str, str]:
     try:
         print("Webhook received")
-        print("GITHUB APP ID: ", GITHUB_APP_ID)
-        payload = await request.body()
-        decoded_data = urllib.parse.unquote(payload.decode())
-        json_data = json.loads(decoded_data)
+        # Validate the webhook signature
+        await github_manager.verify_webhook_signature(request=request, secret=GITHUB_WEBHOOK_SECRET)
+        print("Webhook signature verified")
+
+        # Process the webhook event
+        payload = await request.json()
+        formatted_payload: str = json.dumps(obj=payload, indent=4)
+        print(f"Payload: {formatted_payload}")
 
+        # TODO: Sanitize the payload to remove any sensitive information
         # Handle Create, Delete, and Labeled events
-        await handle_webhook_event(json_data)
+        await handle_webhook_event(payload=payload)
         print("Webhook event handled")
 
         return {"message": "Webhook processed successfully"}
     except Exception as e:
-        print(f"Error: {e}")        
-        raise HTTPException(status_code=500, detail=str(e))
+        print(f"Error: {e}")
+        raise HTTPException(status_code=500, detail=str(object=e))
 
-@app.get("/yo")
-async def root():
+
+@app.get(path="/yo")
+async def root() -> dict[str, str]:
     return {"message": "Hello World"}
 
 

services/github/github_manager.py

@@ -1,45 +1,64 @@
-import datetime
+# Standard imports
 import hashlib  # For HMAC (Hash-based Message Authentication Code) signatures
 import hmac  # For HMAC (Hash-based Message Authentication Code) signatures
-import jwt  # For generating JWTs (JSON Web Tokens)
+import logging
 import requests
+import time
+
+# Third-party imports
+from fastapi import Request
+import jwt  # For generating JWTs (JSON Web Tokens)
 
 
 class GitHubManager:
-    def __init__(self, app_identifier, private_key):
-        self.app_identifier = app_identifier
-        self.private_key = private_key
+    # Constructor to initialize the GitHub App ID and private key to this instance
+    def __init__(self, app_id: str, private_key: bytes) -> None:
+        self.app_id: str = app_id
+        self.private_key: bytes = private_key
 
     # Generate a JWT (JSON Web Token) for GitHub App authentication
-    def create_jwt(self):
-        now = int(datetime.datetime.utcnow().timestamp())
-        payload = {
+    def create_jwt(self) -> str:
+        now = int(time.time())
+        payload: dict[str, int | str] = {
             "iat": now,  # Issued at time
-            "exp": now + (10 * 60),  # JWT expires in 10 minutes
-            "iss": self.app_identifier,  # Issuer
-            "sub": self.app_identifier  # Subject
+            "exp": now + 600,  # JWT expires in 10 minutes
+            "iss": self.app_id,  # Issuer
         }
         # The reason we use RS256 is that GitHub requires it for JWTs
-        return jwt.encode(payload, self.private_key, algorithm="RS256")
+        return jwt.encode(payload=payload, key=self.private_key, algorithm="RS256")
 
     # Verify the webhook signature for security
-    async def verify_webhook_signature(self, request, secret):
-        signature = request.headers.get("X-Hub-Signature-256")
-        body = await request.body()
+    async def verify_webhook_signature(self, request: Request, secret: str) -> None:
+        signature: str | None = request.headers.get("X-Hub-Signature-256")
+        if signature is None:
+            raise ValueError("Missing webhook signature")
+        body: bytes = await request.body()
 
         # Compare the computed signature with the one in the headers
-        expected_signature = "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
+        hmac_key: bytes = secret.encode()
+        hmac_signature: str = hmac.new(key=hmac_key, msg=body, digestmod=hashlib.sha256).hexdigest()
+        expected_signature: str = "sha256=" + hmac_signature
         if not hmac.compare_digest(signature, expected_signature):
             raise ValueError("Invalid webhook signature")
 
     # Get an access token for the installed GitHub App
-    def get_installation_access_token(self, installation_id):
-        jwt_token = self.create_jwt()
-        headers = {
-            "Authorization": f"Bearer {jwt_token}",
-            "Accept": "application/vnd.github.v3+json"
-        }
-        url = f"https://api.github.com/app/installations/{installation_id}/access_tokens"
-        response = requests.post(url, headers=headers)
-        response.raise_for_status()  # Raises HTTPError for bad responses
-        return response.json()["token"]
+    def get_installation_access_token(self, installation_id: int) -> tuple[str, str]:
+        try:
+            jwt_token: str = self.create_jwt()
+            headers: dict[str, str] = {
+                "Authorization": f"Bearer {jwt_token}",
+                "Accept": "application/vnd.github.v3+json",
+                "X-GitHub-Api-Version": "2022-11-28"
+            }
+            url: str = f"https://api.github.com/app/installations/{installation_id}/access_tokens"
+
+            response = requests.post(url=url, headers=headers)
+            response.raise_for_status()  # Raises HTTPError for bad responses
+            json = response.json()
+            return json["token"], json["expires_at"]
+        except requests.exceptions.HTTPError as e:
+            logging.error(msg=f"HTTP Error: {e.response.status_code} - {e.response.text}")
+            raise
+        except Exception as e:
+            logging.error(msg=f"Error: {e}")
+            raise

services/github/github_types.py

@@ -0,0 +1,136 @@
+from typing import TypedDict, Dict, List, Optional
+
+
+class LabelInfo(TypedDict):
+    id: int
+    node_id: str
+    url: str
+    name: str
+    color: str
+    default: bool
+    description: Optional[str]
+
+
+class UserInfo(TypedDict):
+    login: str
+    id: int
+    node_id: str
+    avatar_url: str
+    gravatar_id: str
+    url: str
+    html_url: str
+    followers_url: str
+    following_url: str
+    gists_url: str
+    starred_url: str
+    subscriptions_url: str
+    organizations_url: str
+    repos_url: str
+    events_url: str
+    received_events_url: str
+    type: str
+    site_admin: bool
+
+
+class IssueInfo(TypedDict):
+    url: str
+    repository_url: str
+    labels_url: str
+    comments_url: str
+    events_url: str
+    html_url: str
+    id: int
+    node_id: str
+    number: int
+    title: str
+    user: UserInfo
+    labels: List[LabelInfo]
+    state: str
+    locked: bool
+    assignee: Optional[UserInfo]
+    assignees: List[UserInfo]
+    milestone: Optional[str]
+    comments: int
+    created_at: str
+    updated_at: str
+    closed_at: Optional[str]
+    author_association: str
+    active_lock_reason: Optional[str]
+    body: Optional[str]
+    reactions: Dict[str, int]
+    timeline_url: str
+    performed_via_github_app: Optional[str]
+    state_reason: Optional[str]
+
+
+class OrganizationInfo(TypedDict):
+    login: str
+    id: int
+    node_id: str
+    url: str
+    repos_url: str
+    events_url: str
+    hooks_url: str
+    issues_url: str
+    members_url: str
+    public_members_url: str
+    avatar_url: str
+    description: Optional[str]
+
+
+class RepositoryInfo(TypedDict):
+    id: int
+    node_id: str
+    name: str
+    full_name: str
+    private: bool
+
+
+class PermissionsInfo(TypedDict):
+    actions: str
+    contents: str
+    metadata: str
+    workflows: str
+    repository_hooks: str
+
+
+class InstallationInfo(TypedDict):
+    id: int
+    account: UserInfo
+    repository_selection: str
+    access_tokens_url: str
+    repositories_url: str
+    html_url: str
+    app_id: int
+    app_slug: str
+    target_id: int
+    target_type: str
+    permissions: PermissionsInfo
+    events: List[str]
+    created_at: str
+    updated_at: str
+    single_file_name: Optional[str]
+    has_multiple_single_files: bool
+    single_file_paths: List[str]
+    suspended_by: Optional[str]
+    suspended_at: Optional[str]
+
+
+class GitHubInstallationPayload(TypedDict):
+    action: str
+    installation: InstallationInfo
+    repositories: List[RepositoryInfo]
+    repository_selection: str
+    repositories_added: List[RepositoryInfo]
+    repositories_removed: List[RepositoryInfo]
+    requester: Optional[UserInfo]
+    sender: UserInfo
+
+
+class GitHubLabeledPayload(TypedDict):
+    action: str
+    issue: IssueInfo
+    label: LabelInfo
+    repository: RepositoryInfo
+    organization: OrganizationInfo
+    sender: UserInfo