refactor

pr_agent/servers/github_app.py

@@ -1,3 +1,4 @@
+from typing import Dict, Any
 import logging
 import sys
 
@@ -14,50 +15,65 @@ router = APIRouter()
 
 @router.post("/api/v1/github_webhooks")
 async def handle_github_webhooks(request: Request, response: Response):
-    logging.debug("Received a github webhook")
+    """
+    Receives and processes incoming GitHub webhook requests.
+    Verifies the request signature, parses the request body, and passes it to the handle_request function for further processing.
+    """
+    logging.debug("Received a GitHub webhook")
+    
     try:
         body = await request.json()
     except Exception as e:
         logging.error("Error parsing request body", e)
         raise HTTPException(status_code=400, detail="Error parsing request body") from e
+    
     body_bytes = await request.body()
     signature_header = request.headers.get('x-hub-signature-256', None)
-    try:
+    
-        webhook_secret = settings.github.webhook_secret
+    webhook_secret = getattr(settings.github, 'webhook_secret', None)
-    except AttributeError:
+    
-        webhook_secret = None
     if webhook_secret:
         verify_signature(body_bytes, webhook_secret, signature_header)
+    
     logging.debug(f'Request body:\n{body}')
+    
     return await handle_request(body)
 
 
-async def handle_request(body):
+async def handle_request(body: Dict[str, Any]):
-    action = body.get("action", None)
+    """
-    installation_id = body.get("installation", {}).get("id", None)
+    Handle incoming GitHub webhook requests.
+
+    Args:
+        body: The request body.
+    """
+    action = body.get("action")
+    installation_id = body.get("installation", {}).get("id")
     settings.set("GITHUB.INSTALLATION_ID", installation_id)
     agent = PRAgent()
+
     if action == 'created':
         if "comment" not in body:
             return {}
-        comment_body = body.get("comment", {}).get("body", None)
+        comment_body = body.get("comment", {}).get("body")
-        if 'sender' in body and 'login' in body['sender'] and 'bot' in body['sender']['login']:
+        sender = body.get("sender", {}).get("login")
+        if sender and 'bot' in sender:
             return {}
-        if "issue" not in body and "pull_request" not in body["issue"]:
+        if "issue" not in body or "pull_request" not in body["issue"]:
             return {}
         pull_request = body["issue"]["pull_request"]
-        api_url = pull_request.get("url", None)
+        api_url = pull_request.get("url")
         await agent.handle_request(api_url, comment_body)
 
     elif action in ["opened"] or 'reopened' in action:
-        pull_request = body.get("pull_request", None)
+        pull_request = body.get("pull_request")
         if not pull_request:
             return {}
-        api_url = pull_request.get("url", None)
+        api_url = pull_request.get("url")
-        if api_url is None:
+        if not api_url:
             return {}
         await agent.handle_request(api_url, "/review")
-    else:
+
     return {}
 
 

pr_agent/servers/github_polling.py

@@ -15,28 +15,40 @@ NOTIFICATION_URL = "https://api.github.com/notifications"
 
 
 def now() -> str:
+    """
+    Get the current UTC time in ISO 8601 format.
+    
+    Returns:
+        str: The current UTC time in ISO 8601 format.
+    """
     now_utc = datetime.now(timezone.utc).isoformat()
     now_utc = now_utc.replace("+00:00", "Z")
     return now_utc
 
 
 async def polling_loop():
+    """
+    Polls for notifications and handles them accordingly.
+    """
     handled_ids = set()
     since = [now()]
     last_modified = [None]
     git_provider = get_git_provider()()
     user_id = git_provider.get_user_id()
     agent = PRAgent()
+
     try:
         deployment_type = settings.github.deployment_type
         token = settings.github.user_token
     except AttributeError:
         deployment_type = 'none'
         token = None
+
     if deployment_type != 'user':
         raise ValueError("Deployment mode must be set to 'user' to get notifications")
     if not token:
         raise ValueError("User token must be set to get notifications")
+
     async with aiohttp.ClientSession() as session:
         while True:
             try:
@@ -52,6 +64,7 @@ async def polling_loop():
                     params["since"] = since[0]
                 if last_modified[0]:
                     headers["If-Modified-Since"] = last_modified[0]
+
                 async with session.get(NOTIFICATION_URL, headers=headers, params=params) as response:
                     if response.status == 200:
                         if 'Last-Modified' in response.headers: