2c6298ca4b
- Set up project structure with FastAPI - Implement user and account management - Add send and receive money functionality - Set up transaction processing system - Add JWT authentication - Configure SQLAlchemy with SQLite - Set up Alembic for database migrations - Create comprehensive API documentation
94 lines
2.4 KiB
Python
94 lines
2.4 KiB
Python
from typing import Optional, List
|
|
import random
|
|
import string
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.account import Account
|
|
from app.schemas.account import AccountCreate, AccountUpdate
|
|
|
|
|
|
def generate_account_number() -> str:
|
|
"""
|
|
Generate a random account number
|
|
"""
|
|
# Generate a 10-digit account number
|
|
return ''.join(random.choices(string.digits, k=10))
|
|
|
|
|
|
def get_account_by_id(db: Session, id: int) -> Optional[Account]:
|
|
"""
|
|
Get an account by ID
|
|
"""
|
|
return db.query(Account).filter(Account.id == id).first()
|
|
|
|
|
|
def get_account_by_number(db: Session, account_number: str) -> Optional[Account]:
|
|
"""
|
|
Get an account by account number
|
|
"""
|
|
return db.query(Account).filter(Account.account_number == account_number).first()
|
|
|
|
|
|
def get_user_accounts(db: Session, owner_id: int, skip: int = 0, limit: int = 100) -> List[Account]:
|
|
"""
|
|
Get all accounts owned by a user
|
|
"""
|
|
return db.query(Account).filter(Account.owner_id == owner_id).offset(skip).limit(limit).all()
|
|
|
|
|
|
def create_account(db: Session, account_in: AccountCreate, owner_id: int) -> Account:
|
|
"""
|
|
Create a new account for a user
|
|
"""
|
|
# Generate a unique account number
|
|
account_number = generate_account_number()
|
|
while get_account_by_number(db, account_number):
|
|
account_number = generate_account_number()
|
|
|
|
db_account = Account(
|
|
account_number=account_number,
|
|
owner_id=owner_id,
|
|
account_type=account_in.account_type,
|
|
balance=0.0,
|
|
currency=account_in.currency,
|
|
)
|
|
db.add(db_account)
|
|
db.commit()
|
|
db.refresh(db_account)
|
|
return db_account
|
|
|
|
|
|
def update_account(db: Session, account: Account, account_in: AccountUpdate) -> Account:
|
|
"""
|
|
Update an account
|
|
"""
|
|
update_data = account_in.dict(exclude_unset=True)
|
|
|
|
for field, value in update_data.items():
|
|
setattr(account, field, value)
|
|
|
|
db.add(account)
|
|
db.commit()
|
|
db.refresh(account)
|
|
return account
|
|
|
|
|
|
def delete_account(db: Session, account: Account) -> Account:
|
|
"""
|
|
Delete an account
|
|
"""
|
|
db.delete(account)
|
|
db.commit()
|
|
return account
|
|
|
|
|
|
def update_account_balance(db: Session, account: Account, amount: float) -> Account:
|
|
"""
|
|
Update an account balance
|
|
"""
|
|
account.balance += amount
|
|
db.add(account)
|
|
db.commit()
|
|
db.refresh(account)
|
|
return account |