44 lines
1.7 KiB
Python
44 lines
1.7 KiB
Python
from models.team import Team
|
|
from models.webhook import Webhook
|
|
from models.base import Base
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from urllib.parse import quote_plus
|
|
|
|
class DatabaseManager:
|
|
def __init__(self, db_name: str, db_user: str, db_password: str, db_host: str, db_port: str):
|
|
# URL encode the password to handle special characters
|
|
encoded_password = quote_plus(db_password)
|
|
self.engine = create_engine(f"postgresql://{db_user}:{encoded_password}@{db_host}:{db_port}/{db_name}")
|
|
self.session = sessionmaker(bind=self.engine)
|
|
self.__session = self.session()
|
|
|
|
Base.metadata.create_all(self.engine)
|
|
|
|
def get_session(self):
|
|
return self.session()
|
|
|
|
def get_teams(self) -> list[Team]:
|
|
return self.__session.query(Team).all()
|
|
|
|
def get_team_by_id(self, id: str) -> Team:
|
|
team = self.__session.query(Team).filter(Team.id == id).first()
|
|
if team is None:
|
|
raise ValueError(f"Team with id {id} not found")
|
|
return team
|
|
|
|
def get_team_by_microsoft_team_id(self, microsoft_team_id: str) -> Team:
|
|
team = self.__session.query(Team).filter(Team.microsoft_team_id == microsoft_team_id).first()
|
|
if team is None:
|
|
raise ValueError(f"Team with microsoft team id {microsoft_team_id} not found")
|
|
return team
|
|
|
|
def get_webhooks(self) -> list[Webhook]:
|
|
return self.__session.query(Webhook).all()
|
|
|
|
def get_webhook_by_id(self, id: str) -> Webhook:
|
|
webhook = self.__session.query(Webhook).filter(Webhook.id == id).first()
|
|
if webhook is None:
|
|
raise ValueError(f"Webhook with id {id} not found")
|
|
return webhook |