72 lines
2.3 KiB
Python
72 lines
2.3 KiB
Python
import time
|
|
from typing import Any
|
|
from sqlalchemy.orm import Session
|
|
from app.db.cruddb import domainService,tenantService
|
|
from app.db.session import Database
|
|
|
|
class MemoryCache:
|
|
def __init__(self, max_cache_size: int = 100, ttl: int = 60):
|
|
self.cache = {}
|
|
self.max_cache_size = max_cache_size
|
|
self.ttl = ttl
|
|
|
|
def get(self, key: str) -> Any:
|
|
item = self.cache.get(key)
|
|
if item:
|
|
if time.time() - item['timestamp'] > self.ttl:
|
|
self.cache.pop(key)
|
|
return None
|
|
return item['value']
|
|
return None
|
|
|
|
def set(self, key: str, value: Any) -> None:
|
|
if len(self.cache) >= self.max_cache_size:
|
|
self.cache.pop(next(iter(self.cache)))
|
|
self.cache[key] = {'value': value, 'timestamp': time.time()}
|
|
|
|
# def clear(self,key) -> None:
|
|
# self.cache.pop(key,None)
|
|
|
|
def clear(self) -> None:
|
|
self.cache.clear()
|
|
|
|
|
|
|
|
class domainCache:
|
|
|
|
def __init__(self):
|
|
self.memoryCache = MemoryCache(max_cache_size=50, ttl=120)
|
|
|
|
def set_default_domain(self, db: Session,userid: int,domainid:str):
|
|
domain = domainService.set_default_domain(db,userid,domainid)
|
|
if domain:
|
|
self.memoryCache.set(f"DOMAIN_{userid}",domain.url)
|
|
return domain
|
|
|
|
def get_default_domainurl(self,db: Session, userid: int):
|
|
if not self.memoryCache.get(f"DOMAIN_{userid}"):
|
|
domain = domainService.get_default_domain(db,userid)
|
|
if domain:
|
|
self.memoryCache.set(f"DOMAIN_{userid}",domain.url)
|
|
return self.memoryCache.get(f"DOMAIN_{userid}")
|
|
|
|
def clear_default_domainurl(self):
|
|
self.memoryCache.clear()
|
|
|
|
domainCacheService =domainCache()
|
|
|
|
|
|
class tenantCache:
|
|
|
|
def __init__(self):
|
|
self.memoryCache = MemoryCache(max_cache_size=50, ttl=120)
|
|
|
|
def get_tenant_db(self,db: Session, tenantid: str):
|
|
if not self.memoryCache.get(f"TENANT_{tenantid}"):
|
|
tenant = tenantService.get_tenant(db,tenantid)
|
|
if tenant:
|
|
database = Database(tenant.db)
|
|
self.memoryCache.set(f"TENANT_{tenantid}",database)
|
|
return self.memoryCache.get(f"TENANT_{tenantid}")
|
|
|
|
tenantCacheService =tenantCache() |