from pydantic import BaseModel, ValidationError
from typing import Optional, Dict, Any
import os
from supabase import create_client, Client
supabase_url: str = os.environ.get("XLX_SUPABASE_URL")
supabase_key: str = os.environ.get("XLX_SUPABASE_ANON_KEY")
if not supabase_url or not supabase_key:
print("Debug: Supabase URL or Key is missing in environment variables.")
raise ValueError(
"Supabase URL or Key is missing in environment variables.")
print(f"Debug: Supabase URL - {supabase_url}")
print(f"Debug: Supabase Key - {supabase_key[:5]}... (truncated for security)")
supabase: Client = create_client(supabase_url, supabase_key)
print("Debug: Supabase client initialized successfully.")
class PmDeployment(BaseModel):
project_id: str
deployment_id: str
status: str
pm_id: Optional[int] = None
pid: Optional[int] = None
commit_id: Optional[str] = None
repository_url: Optional[str] = None
commit_comment: Optional[str] = None
time: Optional[int] = None
commit_author_username: Optional[str] = None
xbp_command: Optional[str] = None
github_username: Optional[str] = None
sha: Optional[str] = None
repository_name: Optional[str] = None
sha_short: Optional[str] = None
organization_name: Optional[str] = None
project_name: Optional[str] = None
def create_deployment(self):
try:
print("Debug: Starting create_deployment method.")
deployment_data = self.dict(exclude_unset=True)
print(f"Debug: Deployment data to insert - {deployment_data}")
response = supabase.table("pm_deployments").insert(
deployment_data).execute()
print(f"Debug: Supabase response - {response}")
if response.data:
print("Debug: Deployment successfully created in Supabase.")
else:
print(f"Debug: Failed to create deployment: {response.data}")
return response
except ValidationError as ve:
print(f"Debug: Validation error occurred: {ve}")
raise
except Exception as e:
print(
f"Debug: An error occurred while creating the deployment: {str(e)}"
)
raise
def update_deployment(self, deployment_id: str, update_data: Dict[str,
Any]):
try:
print("Debug: Starting update_deployment method.")
print(f"Debug: Deployment ID - {deployment_id}")
print(f"Debug: Update data - {update_data}")
response = supabase.table("pm_deployments").update(update_data).eq(
"deployment_id", deployment_id).execute()
print(f"Debug: Supabase response - {response}")
if response.data:
print("Debug: Deployment successfully updated in Supabase.")
else:
print(f"Debug: Failed to update deployment: {response.data}")
return response
except ValidationError as ve:
print(f"Debug: Validation error occurred: {ve}")
raise
except Exception as e:
print(
f"Debug: An error occurred while updating the deployment: {str(e)}"
)
raise