from sqlalchemy.orm import Session from ..models.account import Account from ..schemas.account import AccountCreate, AccountUpdate from typing import List, Optional class AccountService: @staticmethod def create_account(db: Session, account_data: AccountCreate) -> Account: """创建账号""" db_account = Account(**account_data.dict()) db.add(db_account) db.commit() db.refresh(db_account) return db_account @staticmethod def get_account(db: Session, account_id: int) -> Optional[Account]: """根据ID获取账号""" return db.query(Account).filter(Account.id == account_id).first() @staticmethod def get_account_by_account(db: Session, account: str) -> Optional[Account]: """根据账号名获取账号""" return db.query(Account).filter(Account.account == account).first() @staticmethod def get_accounts(db: Session, skip: int = 0, limit: int = 100) -> List[Account]: """获取账号列表""" return db.query(Account).offset(skip).limit(limit).all() @staticmethod def update_account(db: Session, account_id: int, account_data: AccountUpdate) -> Optional[Account]: """更新账号""" db_account = db.query(Account).filter(Account.id == account_id).first() if db_account: update_data = account_data.dict(exclude_unset=True) for field, value in update_data.items(): setattr(db_account, field, value) db.commit() db.refresh(db_account) return db_account @staticmethod def delete_account(db: Session, account_id: int) -> bool: """删除账号""" db_account = db.query(Account).filter(Account.id == account_id).first() if db_account: db.delete(db_account) db.commit() return True return False