90 lines
2.2 KiB
Python
90 lines
2.2 KiB
Python
import uuid
|
|
|
|
|
|
class DuplicateInstanceError(Exception):
|
|
def __init__(self, instance):
|
|
self.instance = instance
|
|
|
|
|
|
class BaseInstance:
|
|
"""
|
|
Base class for all instances (manageable by InstancesManager)
|
|
"""
|
|
|
|
def __init__(self, session: dict, _id: str):
|
|
self._session = session
|
|
self._id = _id
|
|
InstancesManager.register(session, self)
|
|
|
|
def get_id(self):
|
|
return self._id
|
|
|
|
def get_session(self):
|
|
return self._session
|
|
|
|
|
|
class SingleInstance(BaseInstance):
|
|
"""
|
|
Base class for instances that can only have one instance at a time.
|
|
"""
|
|
|
|
def __init__(self, session: dict, prefix: str):
|
|
super().__init__(session, prefix)
|
|
self._instance = None
|
|
|
|
|
|
class MultipleInstance(BaseInstance):
|
|
"""
|
|
Base class for instances that can have multiple instances at a time.
|
|
"""
|
|
|
|
def __init__(self, session: dict, prefix: str):
|
|
super().__init__(session, f"{prefix}-{str(uuid.uuid4())}")
|
|
self._instance = None
|
|
|
|
|
|
class InstancesManager:
|
|
instances = {}
|
|
|
|
@staticmethod
|
|
def register(session: dict, instance: BaseInstance):
|
|
"""
|
|
Register an instance in the manager, so that it can be retrieved later.
|
|
:param session:
|
|
:param instance:
|
|
:return:
|
|
"""
|
|
key = (InstancesManager._get_session_id(session), instance.get_id())
|
|
|
|
if isinstance(instance, SingleInstance) and key in InstancesManager.instances:
|
|
raise DuplicateInstanceError(instance)
|
|
|
|
InstancesManager.instances[key] = instance
|
|
return instance
|
|
|
|
@staticmethod
|
|
def get(session: dict, instance_id: str, instance_type: type = None, *args, **kwargs):
|
|
"""
|
|
Get or create an instance of the given type (from its id)
|
|
:param session:
|
|
:param instance_id:
|
|
:param instance_type:
|
|
:param args:
|
|
:param kwargs:
|
|
:return:
|
|
"""
|
|
try:
|
|
key = (InstancesManager._get_session_id(session), instance_id)
|
|
|
|
return InstancesManager.instances[key]
|
|
except KeyError:
|
|
return instance_type(session, *args, **kwargs) # it will be automatically registered
|
|
|
|
@staticmethod
|
|
def _get_session_id(session):
|
|
if not session:
|
|
return "** NOT LOGGED IN **"
|
|
if "user_info" not in session:
|
|
return "** UNKNOWN USER **"
|
|
return session["user_info"].get("id", "** INVALID SESSION **")
|