# Copyright (c) 2024 Ferrosoft GmbH. All rights reserved.
import logging
from abc import ABC, abstractmethod
import requests
from django.core.cache import cache
from django.utils.translation import gettext as _
from ferrosoft.apps.ticketing.models import (
Ticket,
Settings,
TicketType,
TicketPriority,
)
[docs]
class TicketingService(ABC):
[docs]
@abstractmethod
def create_ticket(
self,
ticket: Ticket,
) -> Ticket:
raise NotImplementedError
[docs]
@classmethod
def get_instance(cls) -> "TicketingService":
return TaigaTicketingService.get_instance()
[docs]
class TaigaTicketingService(TicketingService):
def __init__(self, api_client: "TaigaApiClient"):
self.api_client = api_client
[docs]
@classmethod
def get_instance(cls) -> TicketingService:
return cls(api_client=TaigaApiClient.get_instance())
[docs]
def create_ticket(
self,
ticket: Ticket,
) -> Ticket:
settings = Settings.get_instance()
ticket_options = settings.get_ticket_type_options(ticket_type=ticket.type)
if not ticket_options:
raise RuntimeError(
"No project option found for ticket type: %s" % ticket.type
)
project = ticket_options.project
ticket_data = self.api_client.create_ticket(
title=ticket.title,
description=ticket.description,
type=project.get_type_id(TicketType(ticket.type).label),
priority=project.get_priority_id(TicketPriority(ticket.priority).label),
project=project.project_id,
)
ticket.foreign_id = ticket_data["id"]
ticket.save()
self.api_client.add_comment(ticket_data["id"], _("Customer Support Ticket"))
if ticket.attachment:
self.api_client.attach_file(project=project.project_id, ticket=ticket)
return ticket
[docs]
class Config:
AUTH_TOKEN_CACHE_KEY = "taiga_auth_token"
AUTH_TOKEN_EXPIRY = 12 * 60 * 60
def __init__(self, base_url, username, password):
self.base_url = base_url
self.username = username
self.password = password
[docs]
@classmethod
def from_django_settings(cls):
from django.conf import settings
config = getattr(settings, "TAIGA_CONFIG", {})
it = cls(
config.get("BASE_URL"),
config.get("USERNAME"),
config.get("PASSWORD"),
)
if not all([it.base_url, it.username, it.password]):
raise RuntimeError("Taiga settings are missing or empty")
return it
[docs]
def authenticate(self) -> str:
auth_token = cache.get(self.AUTH_TOKEN_CACHE_KEY)
if not auth_token:
auth_token = self._get_auth_token()
cache.set(
self.AUTH_TOKEN_CACHE_KEY,
auth_token,
timeout=self.AUTH_TOKEN_EXPIRY,
)
return auth_token
def _get_auth_token(self) -> str:
response = requests.post(
"%s/auth" % self.base_url,
json={
"username": self.username,
"password": self.password,
"type": "normal",
},
)
if response.status_code == 200:
return response.json().get("auth_token")
error_detail = response.json().get("detail")
logging.error(error_detail)
raise RuntimeError(error_detail)
[docs]
class TaigaApiClient:
def __init__(self, config: Config, auth_token: str):
self.config = config
self.auth_token = auth_token
[docs]
@classmethod
def get_instance(cls) -> "TaigaApiClient":
"""
Create an instance of TaigaApiClient.
This method loads settings, retrieves an auth token (from cache or API), and returns an instance.
"""
config = Config.from_django_settings()
return cls(config, config.authenticate())
def _request_headers(self):
return {
"Authorization": "Bearer %s" % self.auth_token,
"Content-Type": "application/json",
}
[docs]
def create_ticket(
self,
title: str,
description: str,
type: int,
priority: int,
project: int,
) -> dict:
url = "%s/issues" % self.config.base_url
data = {
"subject": title,
"description": description,
"project": project,
"type": type,
"priority": priority,
}
response = requests.post(url, json=data, headers=self._request_headers())
if response.status_code != 201:
logging.error(response.json())
raise RuntimeError("Ticket could not be created")
return response.json()
[docs]
def attach_file(self, project: int, ticket: Ticket) -> None:
headers = self._request_headers()
headers.pop("Content-Type")
requests.post(
url="%s/issues/attachments" % self.config.base_url,
files={
"attached_file": (
ticket.attachment.file.name,
ticket.attachment.file,
)
},
data={
"project": project,
"object_id": ticket.foreign_id,
},
headers=headers,
)