Spaces:
Paused
π Developer Onboarding & Architecture Guide
Welcome to the Zenith Fraud Detection backend! This guide will help you understand our architecture, where to find things, and how to add new features correctly.
ποΈ Architecture Overview
We use a Domain-Driven Design (DDD) approach with a clean separation between Domain Services (business logic) and Infrastructure (technical details).
The Layered Model
βββββββββββββββββββββββββββββββββββββββ
β API Layer β
β (Routers, GraphQL Resolvers, etc.) β
βββββββββββββββββ¬ββββββββββββββββββββββ
β
βββββββββββββββββΌββββββββββββββββββββββ
β Domain Services β
β (Business Logic, Rules, Workflows) β
βββββββββββββββββ¬ββββββββββββββββββββββ
β
βββββββββββββββββΌββββββββββββββββββββββ
β Data Access Layer β
β (Repositories, Models, Schemas) β
βββββββββββββββββββββββββββββββββββββββ
πΊοΈ Service Map
| Service | Located In | Responsibility |
|---|---|---|
| CaseService | app/modules/cases |
Manage fraud cases, transitions, priority |
| TransactionService | app/modules/transactions |
Manage financial transactions, validation |
| UserService | app/modules/users |
User accounts, preferences, roles |
| AnalyticsService | app/modules/analytics |
heavy aggregations, reports, stats |
| EvidenceService | app/modules/evidence |
File uploads, processing, malware scan |
| AuthService | app/modules/auth |
Login, JWT, password hashing |
Infrastructure Services (Pure Technical)
Located in app/services/infrastructure/*:
- DatabaseService: Connection pooling, session management, health checks
- CacheService: Redis caching, invalidation
- StorageService: S3/Local file I/O abstraction
- EmailService: Sending transactional emails
π How to Add a New Feature
Example: Adding a "Flag Transaction" feature
1. Define the Data Access (Repository)
Modify app/modules/transactions/repository.py:
def flag_transaction(self, transaction_id: str, reason: str) -> Transaction:
txn = self.get_by_id(transaction_id)
txn.is_flagged = True
txn.flag_reason = reason
self.session.add(txn)
self.session.flush()
return txn
2. Implement Business Logic (Service)
Modify app/modules/transactions/service.py:
def flag_suspicious_transaction(self, db: Session, txn_id: str, reason: str, user: User) -> Transaction:
# 1. Validate rules
if not user.can_flag_transactions:
raise PermissionError(...)
# 2. Call repository
repo = TransactionRepository(db)
txn = repo.flag_transaction(txn_id, reason)
# 3. Side effects (e.g. notify)
self.notification_service.notify_manager(txn)
return txn
3. Expose via API (Router)
Modify app/modules/transactions/router.py:
@router.post("/{txn_id}/flag")
def flag_transaction_endpoint(
txn_id: str,
reason: FlagRequest,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_active_user)
):
service = TransactionService()
return service.flag_suspicious_transaction(db, txn_id, reason.text, current_user)
π« Common Gotchas
1. db_service vs db Session
- DON'T put business logic in
DatabaseService. - DO use
DatabaseServiceonly to get a session:db = db_service.get_db(). - DO pass the
Sessionobject to domain services.
2. Circular Imports
- DO import types only within
TYPE_CHECKINGblocks if needed. - DO use dependency injection or pass dependencies in methods to avoid import loops.
3. Testing
- DO use
conftest.pyfixtures. - DO mock external services (S3, Email) in unit tests.
- DO use real database (sqlite-memory) for integration tests.
π§ͺ Testing Guidelines
Run all tests:
pytest
Run specific module tests:
pytest tests/unit/modules/test_cases.py
Test Structure
tests/unit/: Fast tests, mocked dependencies.tests/integration/: Slower, real database interactions.tests/performance/: Benchmarks and load tests.
π Troubleshooting
"AttributeError: 'DatabaseService' object has no attribute 'get_user'"
You are trying to call a migrated method on the infrastructure service.
Fix: Use UserService(db).get_user() instead.
"ImportError: cannot import name..."
Likely a circular import.
Fix: Move the import inside the function/method or use typing.TYPE_CHECKING.
Happy Coding! π