Automated Action 2c6298ca4b Implement fintech payment service backend with FastAPI and SQLite
- 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
2025-06-17 11:53:41 +00:00

196 lines
7.0 KiB
Python

from typing import Any, List
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app import crud, models, schemas
from app.api import deps
router = APIRouter()
@router.get("/", response_model=List[schemas.Transaction])
def read_transactions(
skip: int = 0,
limit: int = 100,
current_user: models.User = Depends(deps.get_current_active_user),
db: Session = Depends(deps.get_db),
) -> Any:
"""
Retrieve transactions for the current user.
"""
transactions = crud.get_user_transactions(db, user_id=current_user.id, skip=skip, limit=limit)
return transactions
@router.get("/account/{account_id}", response_model=List[schemas.Transaction])
def read_account_transactions(
account_id: int,
skip: int = 0,
limit: int = 100,
current_user: models.User = Depends(deps.get_current_active_user),
db: Session = Depends(deps.get_db),
) -> Any:
"""
Retrieve transactions for a specific account.
"""
account = crud.get_account_by_id(db, id=account_id)
if not account:
raise HTTPException(status_code=404, detail="Account not found")
if account.owner_id != current_user.id:
raise HTTPException(status_code=403, detail="Not enough permissions")
transactions = crud.get_account_transactions(db, account_id=account_id, skip=skip, limit=limit)
return transactions
@router.post("/deposit", response_model=schemas.Transaction)
def create_deposit(
deposit_in: schemas.DepositCreate,
current_user: models.User = Depends(deps.get_current_active_user),
db: Session = Depends(deps.get_db),
) -> Any:
"""
Create a deposit transaction.
"""
# Verify account ownership
account = crud.get_account_by_id(db, id=deposit_in.account_id)
if not account:
raise HTTPException(status_code=404, detail="Account not found")
if account.owner_id != current_user.id:
raise HTTPException(status_code=403, detail="Not enough permissions")
# Create and process transaction
transaction = crud.create_deposit(db, deposit_in=deposit_in, user_id=current_user.id)
transaction = crud.process_transaction(db, transaction=transaction)
return transaction
@router.post("/withdrawal", response_model=schemas.Transaction)
def create_withdrawal(
withdrawal_in: schemas.WithdrawalCreate,
current_user: models.User = Depends(deps.get_current_active_user),
db: Session = Depends(deps.get_db),
) -> Any:
"""
Create a withdrawal transaction.
"""
# Verify account ownership
account = crud.get_account_by_id(db, id=withdrawal_in.account_id)
if not account:
raise HTTPException(status_code=404, detail="Account not found")
if account.owner_id != current_user.id:
raise HTTPException(status_code=403, detail="Not enough permissions")
# Check sufficient balance
if account.balance < withdrawal_in.amount:
raise HTTPException(status_code=400, detail="Insufficient funds")
# Create and process transaction
transaction = crud.create_withdrawal(db, withdrawal_in=withdrawal_in, user_id=current_user.id)
transaction = crud.process_transaction(db, transaction=transaction)
return transaction
@router.post("/transfer", response_model=schemas.Transaction)
def create_transfer(
transfer_in: schemas.TransferCreate,
current_user: models.User = Depends(deps.get_current_active_user),
db: Session = Depends(deps.get_db),
) -> Any:
"""
Create a transfer transaction between accounts.
"""
# Verify sender account ownership
sender_account = crud.get_account_by_id(db, id=transfer_in.sender_account_id)
if not sender_account:
raise HTTPException(status_code=404, detail="Sender account not found")
if sender_account.owner_id != current_user.id:
raise HTTPException(status_code=403, detail="Not enough permissions")
# Verify receiver account exists
receiver_account = crud.get_account_by_id(db, id=transfer_in.receiver_account_id)
if not receiver_account:
raise HTTPException(status_code=404, detail="Receiver account not found")
# Check sufficient balance
if sender_account.balance < transfer_in.amount:
raise HTTPException(status_code=400, detail="Insufficient funds")
# Create and process transaction
transaction = crud.create_transfer(db, transfer_in=transfer_in, user_id=current_user.id)
transaction = crud.process_transaction(db, transaction=transaction)
return transaction
@router.post("/receive", response_model=schemas.Transaction)
def receive_money(
receive_in: schemas.ReceiveMoneyCreate,
current_user: models.User = Depends(deps.get_current_active_user),
db: Session = Depends(deps.get_db),
) -> Any:
"""
Receive money from an external source.
"""
# Verify account ownership
account = crud.get_account_by_id(db, id=receive_in.receiver_account_id)
if not account:
raise HTTPException(status_code=404, detail="Account not found")
if account.owner_id != current_user.id:
raise HTTPException(status_code=403, detail="Not enough permissions")
# Create and process transaction
transaction = crud.create_receive_money(db, receive_in=receive_in, user_id=current_user.id)
transaction = crud.process_transaction(db, transaction=transaction)
return transaction
@router.post("/send", response_model=schemas.Transaction)
def send_money(
send_in: schemas.SendMoneyCreate,
current_user: models.User = Depends(deps.get_current_active_user),
db: Session = Depends(deps.get_db),
) -> Any:
"""
Send money to an external destination.
"""
# Verify account ownership
account = crud.get_account_by_id(db, id=send_in.sender_account_id)
if not account:
raise HTTPException(status_code=404, detail="Account not found")
if account.owner_id != current_user.id:
raise HTTPException(status_code=403, detail="Not enough permissions")
# Check sufficient balance
if account.balance < send_in.amount:
raise HTTPException(status_code=400, detail="Insufficient funds")
# Create and process transaction
transaction = crud.create_send_money(db, send_in=send_in, user_id=current_user.id)
transaction = crud.process_transaction(db, transaction=transaction)
return transaction
@router.get("/{transaction_id}", response_model=schemas.Transaction)
def read_transaction(
transaction_id: int,
current_user: models.User = Depends(deps.get_current_active_user),
db: Session = Depends(deps.get_db),
) -> Any:
"""
Get transaction by ID.
"""
transaction = crud.get_transaction_by_id(db, id=transaction_id)
if not transaction:
raise HTTPException(status_code=404, detail="Transaction not found")
# Verify user is involved in the transaction
if transaction.sender_id != current_user.id and transaction.receiver_id != current_user.id:
raise HTTPException(status_code=403, detail="Not enough permissions")
return transaction