130 lines
3.2 KiB
Python
130 lines
3.2 KiB
Python
"""
|
|
Gitea Webhook data model
|
|
"""
|
|
|
|
from typing import List, Optional
|
|
from pydantic import BaseModel, Field
|
|
from datetime import datetime
|
|
|
|
|
|
class User(BaseModel):
|
|
"""Gitea user model"""
|
|
id: int
|
|
login: str
|
|
full_name: Optional[str] = None
|
|
email: Optional[str] = None
|
|
username: Optional[str] = None
|
|
|
|
|
|
class Commit(BaseModel):
|
|
"""Git commit model"""
|
|
id: str
|
|
message: str
|
|
url: str
|
|
author: User
|
|
timestamp: Optional[datetime] = None
|
|
|
|
|
|
class Repository(BaseModel):
|
|
"""Git repository model"""
|
|
id: int
|
|
name: str
|
|
owner: User
|
|
full_name: str
|
|
private: bool = False
|
|
clone_url: str
|
|
ssh_url: Optional[str] = None
|
|
html_url: str
|
|
default_branch: str = "main"
|
|
|
|
|
|
class GiteaWebhook(BaseModel):
|
|
"""Gitea Webhook model"""
|
|
secret: Optional[str] = None
|
|
ref: str
|
|
before: str
|
|
after: str
|
|
compare_url: Optional[str] = None
|
|
commits: List[Commit] = Field(default_factory=list)
|
|
repository: Repository
|
|
pusher: User
|
|
|
|
def get_branch_name(self) -> str:
|
|
"""Extract branch name from ref"""
|
|
prefix = "refs/heads/"
|
|
if self.ref.startswith(prefix):
|
|
return self.ref[len(prefix):]
|
|
return self.ref
|
|
|
|
def get_event_id(self) -> str:
|
|
"""Generate unique event ID"""
|
|
return f"{self.repository.full_name}-{self.after}"
|
|
|
|
def get_commit_hash(self) -> str:
|
|
"""Get commit hash"""
|
|
return self.after
|
|
|
|
def get_deduplication_key(self) -> str:
|
|
"""Generate deduplication key"""
|
|
branch = self.get_branch_name()
|
|
return f"{self.after}:{branch}"
|
|
|
|
def is_push_event(self) -> bool:
|
|
"""Determine if it is a push event"""
|
|
return self.ref.startswith("refs/heads/")
|
|
|
|
def is_tag_event(self) -> bool:
|
|
"""Determine if it is a tag event"""
|
|
return self.ref.startswith("refs/tags/")
|
|
|
|
def get_commit_message(self) -> str:
|
|
"""Get commit message"""
|
|
if self.commits:
|
|
return self.commits[0].message
|
|
return ""
|
|
|
|
def get_author_info(self) -> dict:
|
|
"""Get author information"""
|
|
if self.commits:
|
|
author = self.commits[0].author
|
|
return {
|
|
"name": author.full_name or author.login,
|
|
"email": author.email,
|
|
"username": author.login
|
|
}
|
|
return {
|
|
"name": self.pusher.full_name or self.pusher.login,
|
|
"email": self.pusher.email,
|
|
"username": self.pusher.login
|
|
}
|
|
|
|
|
|
class WebhookEvent(BaseModel):
|
|
"""Webhook event model"""
|
|
id: str
|
|
repository: str
|
|
branch: str
|
|
commit_hash: str
|
|
event_type: str
|
|
timestamp: datetime
|
|
payload: dict
|
|
|
|
class Config:
|
|
json_encoders = {
|
|
datetime: lambda v: v.isoformat()
|
|
}
|
|
|
|
|
|
class WebhookResponse(BaseModel):
|
|
"""Webhook response model"""
|
|
success: bool
|
|
message: str
|
|
event_id: Optional[str] = None
|
|
job_name: Optional[str] = None
|
|
environment: Optional[str] = None
|
|
timestamp: datetime = Field(default_factory=datetime.utcnow)
|
|
|
|
class Config:
|
|
json_encoders = {
|
|
datetime: lambda v: v.isoformat()
|
|
} |