74 lines
2.9 KiB
Python
74 lines
2.9 KiB
Python
|
"""Repository for bots table operations."""
|
||
|
|
|
||
|
|
from typing import Dict, Any, Optional
|
||
|
|
|
||
|
|
from ..models import Bot
|
||
|
|
from .base_repository import BaseRepository, DatabaseOperationError
|
||
|
|
|
||
|
|
|
||
|
|
class BotRepository(BaseRepository):
|
||
|
|
"""Repository for bots table operations."""
|
||
|
|
|
||
|
|
def add(self, bot_data: Dict[str, Any]) -> Bot:
|
||
|
|
"""Add a new bot to the database."""
|
||
|
|
try:
|
||
|
|
with self.get_session() as session:
|
||
|
|
new_bot = Bot(**bot_data)
|
||
|
|
session.add(new_bot)
|
||
|
|
session.commit()
|
||
|
|
session.refresh(new_bot)
|
||
|
|
self.log_info(f"Added new bot: {new_bot.name}")
|
||
|
|
return new_bot
|
||
|
|
except Exception as e:
|
||
|
|
self.log_error(f"Error adding bot: {e}")
|
||
|
|
raise DatabaseOperationError(f"Failed to add bot: {e}")
|
||
|
|
|
||
|
|
def get_by_id(self, bot_id: int) -> Optional[Bot]:
|
||
|
|
"""Get a bot by its ID."""
|
||
|
|
try:
|
||
|
|
with self.get_session() as session:
|
||
|
|
return session.query(Bot).filter(Bot.id == bot_id).first()
|
||
|
|
except Exception as e:
|
||
|
|
self.log_error(f"Error getting bot by ID {bot_id}: {e}")
|
||
|
|
raise DatabaseOperationError(f"Failed to get bot by ID: {e}")
|
||
|
|
|
||
|
|
def get_by_name(self, name: str) -> Optional[Bot]:
|
||
|
|
"""Get a bot by its name."""
|
||
|
|
try:
|
||
|
|
with self.get_session() as session:
|
||
|
|
return session.query(Bot).filter(Bot.name == name).first()
|
||
|
|
except Exception as e:
|
||
|
|
self.log_error(f"Error getting bot by name {name}: {e}")
|
||
|
|
raise DatabaseOperationError(f"Failed to get bot by name: {e}")
|
||
|
|
|
||
|
|
def update(self, bot_id: int, update_data: Dict[str, Any]) -> Optional[Bot]:
|
||
|
|
"""Update a bot's information."""
|
||
|
|
try:
|
||
|
|
with self.get_session() as session:
|
||
|
|
bot = session.query(Bot).filter(Bot.id == bot_id).first()
|
||
|
|
if bot:
|
||
|
|
for key, value in update_data.items():
|
||
|
|
setattr(bot, key, value)
|
||
|
|
session.commit()
|
||
|
|
session.refresh(bot)
|
||
|
|
self.log_info(f"Updated bot {bot_id}")
|
||
|
|
return bot
|
||
|
|
return None
|
||
|
|
except Exception as e:
|
||
|
|
self.log_error(f"Error updating bot {bot_id}: {e}")
|
||
|
|
raise DatabaseOperationError(f"Failed to update bot: {e}")
|
||
|
|
|
||
|
|
def delete(self, bot_id: int) -> bool:
|
||
|
|
"""Delete a bot by its ID."""
|
||
|
|
try:
|
||
|
|
with self.get_session() as session:
|
||
|
|
bot = session.query(Bot).filter(Bot.id == bot_id).first()
|
||
|
|
if bot:
|
||
|
|
session.delete(bot)
|
||
|
|
session.commit()
|
||
|
|
self.log_info(f"Deleted bot {bot_id}")
|
||
|
|
return True
|
||
|
|
return False
|
||
|
|
except Exception as e:
|
||
|
|
self.log_error(f"Error deleting bot {bot_id}: {e}")
|
||
|
|
raise DatabaseOperationError(f"Failed to delete bot: {e}")
|