Task: gitlab.create-issue
Goal: Open an issue to report experiencing "OSError: [Errno 98] Address already in use" during executions in {{aem-hacker}}.
Task configuration:
issue: report experiencing "OSError: [Errno 98] Address already in use" during executions
search_keywords: OSError Address already in use
namespace: 0ang3el
project_name: aem-hacker

Validation code:
from typing import Any, Dict, List
from urllib.parse import quote

import requests
import json
import os

def _get_headers():
    return {
        "Content-Type": "application/json",
        "Accept": "application/json",
        **json.loads(os.getenv("GITLAB_EXTRA_HTTP_HEADERS", "{}")),
    }

def _get_current_username():
    url = os.getenv("GITLAB_BASE_URL", "").rstrip("/") + "/api/v4/user"
    response = requests.get(url=url, headers=_get_headers())
    if response.status_code == 200:
        return response.json()["username"], response.json()["name"]
    return None, None

def validate(messages: List[Dict[str, str]], config: Dict[str, Any]) -> int:

    # Check whether we have the necessary information to validate
    if not "namespace" in config or not "project_name" in config:
        raise ValueError("Target must contain 'namespace' and 'project_name'")

    # check whether we have messages
    if not messages:
        raise ValueError("Messages are empty")

    project_id = quote(f"{config['namespace']}/{config['project_name']}", safe="")

    if "issue_title" in config:
        search_query = config["issue_title"]
    elif "search_keywords" in config:
        search_query = config["search_keywords"]
    else:
        raise ValueError("Target must contain 'issue_title' or 'search_keywords'")

    url = (
        os.getenv("GITLAB_BASE_URL", "").rstrip("/")
        + "/api/v4/projects/"
        + project_id
        + "/issues?search="
        + quote(search_query)
        + "&in=title"
    )

    # api call
    response = requests.get(url=url, headers=_get_headers())
    response.raise_for_status()
    result = response.json()
    if len(result) == 0:
        return 0
    issue = result[0]

    # check for assignees
    if "assignee" in config:
        assignee_value = config["assignee"].lower().strip()

        # Handle "myself" by resolving to current user
        if assignee_value == "myself":
            username, name = _get_current_username()
            if username is None:
                return 0
            # Check if current user is among assignees
            found = False
            for assignee in issue["assignees"]:
                if (username.lower() in assignee["username"].lower()
                    or name.lower() in assignee["name"].lower()):
                    found = True
                    break
            if not found:
                return 0
        else:
            # Check if the specified assignee is among the issue's assignees
            found = False
            for assignee in issue["assignees"]:
                if (assignee_value in assignee["name"].lower()
                    or assignee_value in assignee["username"].lower()):
                    found = True
                    break
            if not found:
                return 0

    # check for due date
    if "end_date" in config:
        if config["end_date"] != issue["due_date"]:
            return 0

    return 1