103 lines
3.2 KiB
Python
103 lines
3.2 KiB
Python
# common/utils/cache/license_cache.py
|
|
from typing import Dict, Any, Optional
|
|
from datetime import datetime as dt, timezone as tz
|
|
|
|
from flask import current_app
|
|
from sqlalchemy import and_
|
|
from sqlalchemy.inspection import inspect
|
|
|
|
from common.utils.cache.base import CacheHandler
|
|
from common.models.entitlements import License
|
|
|
|
|
|
class LicenseCacheHandler(CacheHandler[License]):
|
|
"""Handles caching of active licenses for tenants"""
|
|
handler_name = 'license_cache'
|
|
|
|
def __init__(self, region):
|
|
super().__init__(region, 'active_license')
|
|
self.configure_keys('tenant_id')
|
|
|
|
def _to_cache_data(self, instance: License) -> Dict[str, Any]:
|
|
"""Convert License instance to cache data using SQLAlchemy inspection"""
|
|
if not instance:
|
|
return {}
|
|
|
|
# Get all column attributes from the SQLAlchemy model
|
|
mapper = inspect(License)
|
|
data = {}
|
|
|
|
for column in mapper.columns:
|
|
value = getattr(instance, column.name)
|
|
|
|
# Handle date serialization
|
|
if isinstance(value, dt):
|
|
data[column.name] = value.isoformat()
|
|
else:
|
|
data[column.name] = value
|
|
|
|
return data
|
|
|
|
def _from_cache_data(self, data: Dict[str, Any], **kwargs) -> License:
|
|
"""Create License instance from cache data using SQLAlchemy inspection"""
|
|
if not data:
|
|
return None
|
|
|
|
# Create a new License instance
|
|
license = License()
|
|
mapper = inspect(License)
|
|
|
|
# Set all attributes dynamically
|
|
for column in mapper.columns:
|
|
if column.name in data:
|
|
value = data[column.name]
|
|
|
|
# Handle date deserialization
|
|
if column.name.endswith('_date') and value:
|
|
if isinstance(value, str):
|
|
value = dt.fromisoformat(value).date()
|
|
|
|
setattr(license, column.name, value)
|
|
|
|
return license
|
|
|
|
def _should_cache(self, value: License) -> bool:
|
|
"""Validate if the license should be cached"""
|
|
return value is not None and value.id is not None
|
|
|
|
def get_active_license(self, tenant_id: int) -> Optional[License]:
|
|
"""
|
|
Get the currently active license for a tenant
|
|
|
|
Args:
|
|
tenant_id: ID of the tenant
|
|
|
|
Returns:
|
|
License instance if found, None otherwise
|
|
"""
|
|
|
|
def creator_func(tenant_id: int) -> Optional[License]:
|
|
from common.extensions import db
|
|
current_date = dt.now(tz=tz.utc).date()
|
|
|
|
# TODO --> Active License via active Period?
|
|
|
|
return (db.session.query(License)
|
|
.filter_by(tenant_id=tenant_id)
|
|
.filter(License.start_date <= current_date)
|
|
.last())
|
|
|
|
return self.get(creator_func, tenant_id=tenant_id)
|
|
|
|
def invalidate_tenant_license(self, tenant_id: int):
|
|
"""Invalidate cached license for specific tenant"""
|
|
self.invalidate(tenant_id=tenant_id)
|
|
|
|
|
|
def register_license_cache_handlers(cache_manager) -> None:
|
|
"""Register license cache handlers with cache manager"""
|
|
cache_manager.register_handler(
|
|
LicenseCacheHandler,
|
|
'eveai_model' # Use existing eveai_model region
|
|
)
|