60 lines
1.9 KiB
Python
60 lines
1.9 KiB
Python
"""Integration model."""
|
|
from datetime import datetime
|
|
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, Enum, Boolean, Text, JSON
|
|
from sqlalchemy.orm import relationship
|
|
from app.core.database import Base
|
|
import enum
|
|
|
|
class IntegrationType(str, enum.Enum):
|
|
JIRA_CLOUD = "jira_cloud"
|
|
JIRA_SERVER = "jira_server"
|
|
SERVICENOW = "servicenow"
|
|
ZENDESK = "zendesk"
|
|
GITHUB = "github"
|
|
GITLAB = "gitlab"
|
|
AZURE_DEVOPS = "azure_devops"
|
|
TICKETHUB = "tickethub"
|
|
GITEA = "gitea"
|
|
CUSTOM_WEBHOOK = "custom_webhook"
|
|
|
|
class IntegrationStatus(str, enum.Enum):
|
|
ACTIVE = "active"
|
|
INACTIVE = "inactive"
|
|
ERROR = "error"
|
|
|
|
class Integration(Base):
|
|
__tablename__ = "integrations"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
organization_id = Column(Integer, ForeignKey("organizations.id"), nullable=False)
|
|
|
|
name = Column(String(255), nullable=False)
|
|
type = Column(Enum(IntegrationType), nullable=False)
|
|
status = Column(Enum(IntegrationStatus), default=IntegrationStatus.ACTIVE)
|
|
|
|
# Config
|
|
base_url = Column(String(1024))
|
|
api_key = Column(Text) # Encrypted
|
|
oauth_token = Column(Text)
|
|
webhook_secret = Column(String(255))
|
|
callback_url = Column(String(1024))
|
|
config = Column(JSON, default=dict) # Additional config as JSON
|
|
|
|
# Stats
|
|
issues_processed = Column(Integer, default=0)
|
|
last_sync_at = Column(DateTime)
|
|
last_error = Column(Text)
|
|
|
|
# Settings
|
|
auto_analyze = Column(Boolean, default=True)
|
|
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
|
|
# Relations
|
|
organization = relationship("Organization", back_populates="integrations")
|
|
issues = relationship("Issue", back_populates="integration")
|
|
|
|
@property
|
|
def webhook_url(self) -> str:
|
|
return f"https://jira-fixer.startdata.com.br/api/webhook/{self.organization_id}/{self.type.value}"
|