98 lines
3.4 KiB
Python
98 lines
3.4 KiB
Python
from sqlalchemy import func
|
|
|
|
from models import Account
|
|
from database import DatabaseManager
|
|
from database.exceptions import DatabaseError
|
|
from database.exception_catcher_decorator import handle_database_errors
|
|
from utils.constants import MIN_ACCOUNT_NUMBER, MAX_ACCOUNT_NUMBER
|
|
|
|
|
|
@handle_database_errors
|
|
def get_next_id() -> int:
|
|
with DatabaseManager.get_session() as session:
|
|
current_max_id = session.query(func.max(Account.account_number)).scalar()
|
|
current_max_id = current_max_id + 1 if current_max_id is not None else MIN_ACCOUNT_NUMBER
|
|
|
|
if current_max_id > MAX_ACCOUNT_NUMBER:
|
|
raise DatabaseError("Too many users already exist, cannot open new account", DatabaseError.OUT_OF_ACCOUNT_SPACE)
|
|
|
|
return current_max_id
|
|
|
|
|
|
@handle_database_errors
|
|
def create_account() -> int:
|
|
new_id = get_next_id()
|
|
|
|
with DatabaseManager.get_session() as session:
|
|
new_account = Account(account_number=new_id, balance=0)
|
|
session.add(new_account)
|
|
session.commit()
|
|
return new_id
|
|
|
|
|
|
@handle_database_errors
|
|
def get_account_balance(account_number: int) -> int:
|
|
with DatabaseManager.get_session() as session:
|
|
account: Account = session.query(Account).where(Account.account_number == account_number).one_or_none()
|
|
if account is None:
|
|
raise DatabaseError(f"Account with number {account_number} doesn't exist", DatabaseError.NONEXISTENT_ACCOUNT)
|
|
|
|
return account.balance
|
|
|
|
|
|
@handle_database_errors
|
|
def withdraw_from_account(account_number: int, amount: int):
|
|
modify_balance(account_number, amount, False)
|
|
|
|
|
|
@handle_database_errors
|
|
def deposit_into_account(account_number: int, amount: int):
|
|
modify_balance(account_number, amount, True)
|
|
|
|
|
|
@handle_database_errors
|
|
def modify_balance(account_number: int, amount: int, add: bool):
|
|
with DatabaseManager.get_session() as session:
|
|
account: Account = session.query(Account).where(Account.account_number == account_number).one_or_none()
|
|
if account is None:
|
|
raise DatabaseError(f"Account with number {account_number} doesn't exist", DatabaseError.NONEXISTENT_ACCOUNT)
|
|
|
|
if add:
|
|
account.balance += amount
|
|
else:
|
|
if account.balance - amount < 0:
|
|
raise DatabaseError("Not enough funds on account to withdraw this much", DatabaseError.INSUFFICIENT_BALANCE)
|
|
account.balance -= amount
|
|
|
|
session.commit()
|
|
|
|
|
|
@handle_database_errors
|
|
def delete_account(account_number: int):
|
|
with DatabaseManager.get_session() as session:
|
|
account: Account = session.query(Account).where(Account.account_number == account_number).one_or_none()
|
|
if account is None:
|
|
raise DatabaseError(f"Account with number {account_number} doesn't exist", DatabaseError.NONEXISTENT_ACCOUNT)
|
|
|
|
if account.balance > 0:
|
|
raise DatabaseError("Cannot delete an account with leftover funds", DatabaseError.INVALID_OPERATION)
|
|
|
|
session.delete(account)
|
|
session.commit()
|
|
|
|
|
|
@handle_database_errors
|
|
def get_total_balance() -> int:
|
|
with DatabaseManager.get_session() as session:
|
|
total_sum = session.query(func.sum(Account.balance)).scalar()
|
|
|
|
return total_sum if total_sum is not None else 0
|
|
|
|
|
|
@handle_database_errors
|
|
def get_account_count() -> int:
|
|
with DatabaseManager.get_session() as session:
|
|
total_sum = session.query(func.count(Account.account_number)).scalar()
|
|
|
|
return total_sum if total_sum is not None else 0
|