roboluke-tasks/app/utils/n8n.py

"""N8N utils module."""

import requests

from app.utils.config import config

if config.sentry_enabled:
    import sentry_sdk


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.
    """
    data: dict = {
        "requestor": requestor,
        "title": summary,
        "description": description,
        "completiondate": completion_date,
    }
    if not config.sentry_enabled:
        _data = __n8n_post(data=data)
        return _data
    with sentry_sdk.start_transaction(name="submit_task"):
        _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.
    """
    headers: dict = {"Content-Type": "application/json"}
    if not config.sentry_enabled:
        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
    with sentry_sdk.start_transaction(name="get_tasks"):
        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