Odoo task flow automation using Github events

automated_action.py

# Map task stages to ids
task_stages = {
  "Done": 3
}
activity_types = {
  "Merge PR": 4 
}

for task in records:
  task_is_done = task.stage_id and task.stage_id.id == task_stages["Done"]
  if task.pr_state == "merged" and not task_is_done:
    
    # Update stage
    task.write({"stage_id": task_stages["Done"]})
    
    # Mark all "Merge PR" activities as done
    for activity in task.activity_ids:
      if activity.activity_type_id.id == activity_types["Merge PR"]:
        activity.action_done()

endpoint.py

# Hardcoded secret "Oh no! Anyway.."
github_secret = "SuchMuchToken123ohno!"

# Default result
result = {"response": Response("No Github related actions found", 404)}

# Supported events
supported_events = ["ping", "pull_request"]

# Task obj
Tasks = env["project.task"]

# Task states. Map ids to names
task_stages = {"Review": 35}

# Check signature from request
signature = request.httprequest.headers.get('X-Hub-Signature-256')
raw_data = request.httprequest.data
expected_signature = 'sha256=' + hmac.new(github_secret.encode(), raw_data, hashlib.sha256).hexdigest()

if not hmac.compare_digest(expected_signature, signature):
  result = {"response": Response("Token not such much", 403)}

else:        

  # Check event type
  event_type = request.httprequest.headers.get('X-GitHub-Event')
  
  # Needed to ensure that webhook is working
  if event_type == "ping":
    result = {"response": Response("Odoo webhook is set!")}
  
  elif event_type not in supported_events:
    result = {"response": Response("This event is not supported", 404)}
  
  # Event is supported
  else:
    # Parse the incoming JSON payload.
    payload = request.httprequest.json
    
    # Check if the payload contains a pull_request key.
    pr = payload.get('pull_request')
    if pr:
      # Search task by PR URI
      task = Tasks.search([("pr_uri", "=", pr.get("html_url"))])

      # Task is found
      if task:

        # Will update all at once
        new_task_vals = {}
        
        # -------------------------
        # Actions based on PR state
        # -------------------------
        
        # Extract the pull request status (e.g., "open" or "closed").
        pr_state = pr.get('state')
        
        # Check for merged and draft PRs
        if pr.get('merged'):
          pr_state = "merged"
        elif pr.get("draft"):
          pr_state = "draft"
                
        # Update task status
        if task.pr_state != pr_state:
          task.write({"pr_state": pr_state})
        
        # Extract all label names from the pull request.
        labels = [label.get('name') for label in pr.get('labels', [])]
        
        # -----------------------
        # Actions based on labels
        # -----------------------
        
        # -- Approved -> "Ready for next stage" if task is in "Review"
        
        task_in_review = task.stage_id and task.stage_id.id == task_stages.get("Review")
        if "approved" in labels:
          if task.kanban_state != "done" and task_in_review:
            new_task_vals.update({"kanban_state": "done"})
        else:
          if task.kanban_state == "done":
            new_task_vals.update({"kanban_state": "normal"})
          
        # ----------------------
        # Update task with vals
        # ----------------------
        if new_task_vals:
          task.write(new_task_vals)

        # Return "OK"  
        result = {"response": Response("ok")}