roboluke-tasks/app/utils/n8n.py

import requests
import sentry_sdk

from app.utils.config import config


def __n8n_post(data: dict) -> bool:
    """Post data to N8N webhook URL.
    
    Args:
        data (dict): Data to post to webhook URL.
    
    Returns:
        bool: True if successful, else False.
    """
    headers: dict = {"Content-Type": "application/json"}
    resp: requests.Response = requests.post(
        url=config.n8n_webhook_url,
        headers=headers,
        json=data,
        timeout=10,
        verify=False,
    )
    return bool(resp.status_code == 200)


def submit_task(summary, description, completion_date, requestor) -> bool:
    """Submit task to N8N webhook URL.
    
    Args:
        summary (str): Summary of task.
        description (str): Description of task.
        completion_date (str): Completion date of task.
        requestor (str): Requestor of task.
    
    Returns:
        bool: True if successful, else False.
    """
    with sentry_sdk.start_transaction(name="submit_task"):
        data: dict = {
            "requestor": requestor,
            "title": summary,
            "description": description,
            "completiondate": completion_date,
        }
        _data = __n8n_post(data=data)
    return _data


def get_tasks(requestor) -> bool:
    """Get tasks from N8N webhook URL.
    
    Args:
        requestor (str): Requestor of tasks.
    
    Returns:
        bool: True if successful, else False.
    """
    with sentry_sdk.start_transaction(name="get_tasks"):
        headers: dict = {"Content-Type": "application/json"}
        resp: requests.Response = requests.get(
            url=config.n8n_webhook_url,
            headers=headers,
            timeout=10,
            verify=False,
            params={"requestor": requestor},
        )
        _data = bool(resp.status_code == 200)
    return _data