ServicesCrelio AppArchitectureApp Modules core
Shared infrastructure, base models, utilities, and service clients
- BaseModel foundation with lifecycle hooks
- Utility clients for external services (ES, S3, Redis, Slack, Pusher, DocumentDB)
- Middleware for authentication, sessions, and request handling
- Decorators for common view patterns
- Shared utilities (date, encryption, translation, PDF, image processing)
- Django framework
- External service SDKs (boto3, elasticsearch, redis-py)
- Domain-specific business logic (belongs in domain apps)
- Feature-specific models (belongs in domain apps)
- API endpoints for specific features
| Model | Responsibility | Key Fields | Relations |
|---|
BaseModel | Abstract base with lifecycle hooks | cache_type, allow_individual_instance_caching, webhook_enabled | Inherited by all domain models |
ActivityLogBase | Mixin for activity logging | category_id_mapper | Mixed into BaseModel |
DocumentDBConnection | DocumentDB connection config | account_id, connection_url | None |
| Method | Purpose |
|---|
validate(*args, **kwargs) | Override for input validation |
before_save(*args, **kwargs) | Override for pre-save logic |
after_save(*args, **kwargs) | Override for post-save side effects |
save(*args, **kwargs) | Orchestrates validation → before_save → DB save → after_save |
set_values(values: dict) | Bulk set attributes from dictionary |
validate_mandatory_fields() | Check required fields are present |
get(ExceptionToRaise, **filters) | Fetch single instance or raise |
get_cached_instance(instance_id, ...) | Fetch from Redis cache |
notify_on_slack(channel, message, ...) | Send Slack notification |
| Method | Purpose |
|---|
add_activity_log(action, message, session, ...) | Log activity to Elasticsearch |
get_activities(instance_id, start_date, end_date, ...) | Retrieve activity logs |
prepare_activity_log_payload(action, session) | Override to customize log payload |
The core app provides infrastructure mixins, not domain logic. Models in domain apps inherit from BaseModel to get:
- Consistent lifecycle hooks
- Activity logging capability
- Caching utilities
- Slack notifications
save() always calls validate() → before_save() → super().save() → after_save()
- If
validate() raises ValidationError, save is aborted
after_save() runs INSIDE the transaction (careful with external calls)
- Format:
{ModelName}_CentreId{lab_id}_List or custom cache_key
- TTL:
default_cache_ttl = 7 * 24 * 60 * 60 (7 days)
# Redis hash-based caching
instance = cache.hget(cls.cache_key, instance_id)
if not instance:
instance = cls.get(pk=instance_id)
cache.hset(cls.cache_key, instance_id, json.dumps(serialized))
# core/utils/clients.py
client = get_client("es") # ElasticSearchClient
client = get_client("s3") # S3Client
client = get_client("slack") # SlackClient
client = get_client("fusion") # FusionClient
client = get_client("docdb", account_id=123) # DocumentDBClient
The core app exposes minimal endpoints:
| Endpoint | View | Purpose |
|---|
/core/pdf/ | pdf_view.py | PDF generation utilities |
/core/view/ | view.py | Health checks |
# core/view.py
class GenericView(View):
"""Base view with session utilities"""
def get_lab_id_from_session(self, session):
"""Extract lab_id from session"""
return session.get("labId") or session.get("docLabId")
def get_child_lab_ids(self, parent_lab_id):
"""Get child labs for multi-tenant queries"""
return LabRelations.get_child_labs(parent_lab_id)
| Service | Client Location | Purpose |
|---|
| Elasticsearch | core/utils/elastic_search/ | Activity logs, search |
| Redis Cluster | core/cache.py | Caching, sessions |
| AWS S3 | core/utils/aws/client.py | File storage |
| Slack | core/utils/slack.py | Notifications |
| Pusher | core/utils/pusher/ | Real-time updates |
| DocumentDB | core/utils/documentdb/ | Integration logs |
| Fusion Worker | core/utils/fusion/ | Async job queue |
# core/utils/clients.py
def get_client(service: str, account_id=None, ...):
if service == "es":
return ElasticSearchClient(config=settings.ES_CONFIG)
elif service == "s3":
return S3Client()
elif service == "slack":
return SlackClient()
# ... more clients
The core app does NOT use signals.py. All side effects are explicit in model methods.
| From | To | Type |
|---|
core/models/base.py | core/utils/clients.py | Import for Slack |
core/models/activity_log_base.py | core/utils/activity_log/ | Activity logging |
core/view.py | admin.account.proxies.lab_relation | Lab hierarchy |
cache.hget() / cache.hset() - Redis operations per request
get_client("es") - Client instantiation (consider pooling)
- Model instances cached in Redis hashes by
(model_key, instance_id)
- TTL is 7 days by default
- Manual invalidation via
reset_cache_instance(instance_id)
- Create client in
core/utils/{service_name}/client.py
- Register in
core/utils/clients.py:
elif service == "new_service":
return NewServiceClient(**args)
- Add settings in
config/settings/
- Create mixin in
core/models/{mixin_name}.py
- Add to
BaseModel's parent classes if universally needed
- Or inherit in specific domain models
- Keep utilities stateless
- Use dependency injection via
get_client()
- Log errors to Sentry
- Domain logic in
core/
- Direct database queries without going through models
- Synchronous external calls in hot paths