51 lines
1.9 KiB
Python
51 lines
1.9 KiB
Python
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 |