Refactored instances management

This commit is contained in:
2025-11-23 19:52:03 +01:00
parent 97247f824c
commit b1be747101
24 changed files with 783 additions and 216 deletions

View File

@@ -1,7 +1,8 @@
import uuid
from typing import Self
from typing import Optional
from myfasthtml.controls.helpers import Ids
from myfasthtml.core.utils import pascal_to_snake
special_session = {
"user_info": {"id": "** SPECIAL SESSION **"}
@@ -18,25 +19,76 @@ class BaseInstance:
Base class for all instances (manageable by InstancesManager)
"""
def __init__(self, session: dict, prefix: str, _id: str, parent: Self, auto_register: bool = True):
self._session = session
self._id = _id
self._prefix = prefix
def __new__(cls, *args, **kwargs):
# Extract arguments from both positional and keyword arguments
# Signature matches __init__: parent, session=None, _id=None, auto_register=True
parent = args[0] if len(args) > 0 and isinstance(args[0], BaseInstance) else kwargs.get("parent", None)
session = args[1] if len(args) > 1 and isinstance(args[1], dict) else kwargs.get("session", None)
_id = args[2] if len(args) > 2 and isinstance(args[2], str) else kwargs.get("_id", None)
# Compute _id if not provided
if _id is None:
_id = cls.compute_id()
if session is None:
if parent is not None:
session = parent.get_session()
else:
raise TypeError("Either session or parent must be provided")
session_id = InstancesManager.get_session_id(session)
key = (session_id, _id)
if key in InstancesManager.instances:
res = InstancesManager.instances[key]
if type(res) is not cls:
raise TypeError(f"Instance with id {_id} already exists, but is of type {type(res)}")
return res
# Otherwise create a new instance
instance = super().__new__(cls)
instance._is_new_instance = True # mark as fresh
return instance
def __init__(self, parent: Optional['BaseInstance'],
session: Optional[dict] = None,
_id: Optional[str] = None,
auto_register: bool = True):
if not getattr(self, "_is_new_instance", False):
# Skip __init__ if instance already existed
return
else:
# make sure that it's no longer considered as a new instance
self._is_new_instance = False
self._parent = parent
self._session = session or (parent.get_session() if parent else None)
self._id = _id or self.compute_id()
if auto_register:
InstancesManager.register(session, self)
InstancesManager.register(self._session, self)
def get_id(self):
return self._id
def get_session(self):
def get_session(self) -> dict:
return self._session
def get_prefix(self):
return self._prefix
def get_id(self) -> str:
return self._id
def get_parent(self):
def get_parent(self) -> Optional['BaseInstance']:
return self._parent
@classmethod
def get_prefix(cls):
return f"mf-{pascal_to_snake(cls.__name__)}"
@classmethod
def compute_id(cls):
prefix = cls.get_prefix()
if issubclass(cls, SingleInstance):
_id = prefix
else:
_id = f"{prefix}-{str(uuid.uuid4())}"
return _id
class SingleInstance(BaseInstance):
@@ -44,19 +96,12 @@ class SingleInstance(BaseInstance):
Base class for instances that can only have one instance at a time.
"""
def __init__(self, session: dict, prefix: str, parent, auto_register: bool = True):
super().__init__(session, prefix, prefix, parent, auto_register)
class UniqueInstance(BaseInstance):
"""
Base class for instances that can only have one instance at a time.
Does not throw exception if the instance already exists, it simply overwrites it.
"""
def __init__(self, prefix: str, parent: BaseInstance, auto_register: bool = True):
super().__init__(parent.get_session(), prefix, prefix, parent, auto_register)
self._prefix = prefix
def __init__(self,
parent: Optional[BaseInstance] = None,
session: Optional[dict] = None,
_id: Optional[str] = None,
auto_register: bool = True):
super().__init__(parent, session, _id, auto_register)
class MultipleInstance(BaseInstance):
@@ -64,9 +109,11 @@ class MultipleInstance(BaseInstance):
Base class for instances that can have multiple instances at a time.
"""
def __init__(self, prefix: str, parent: BaseInstance, auto_register: bool = True, _id=None):
super().__init__(parent.get_session(), prefix, _id or f"{prefix}-{str(uuid.uuid4())}", parent, auto_register)
self._prefix = prefix
def __init__(self, parent: BaseInstance,
session: Optional[dict] = None,
_id: Optional[str] = None,
auto_register: bool = True):
super().__init__(parent, session, _id, auto_register)
class InstancesManager:
@@ -80,7 +127,7 @@ class InstancesManager:
:param instance:
:return:
"""
key = (InstancesManager._get_session_id(session), instance.get_id())
key = (InstancesManager.get_session_id(session), instance.get_id())
if isinstance(instance, SingleInstance) and key in InstancesManager.instances:
raise DuplicateInstanceError(instance)
@@ -89,48 +136,27 @@ class InstancesManager:
return instance
@staticmethod
def get(session: dict, instance_id: str, instance_type: type = None, parent: BaseInstance = None, *args, **kwargs):
def get(session: dict, instance_id: str):
"""
Get or create an instance of the given type (from its id)
:param session:
:param instance_id:
:param instance_type:
:param parent:
:param args:
:param kwargs:
:return:
"""
try:
key = (InstancesManager._get_session_id(session), instance_id)
return InstancesManager.instances[key]
except KeyError:
if instance_type:
if not issubclass(instance_type, SingleInstance):
assert parent is not None, "Parent instance must be provided if not SingleInstance"
if isinstance(parent, MultipleInstance):
return instance_type(parent, _id=instance_id, *args, **kwargs)
else:
return instance_type(session, parent=parent, *args, **kwargs) # it will be automatically registered
else:
raise
key = (InstancesManager.get_session_id(session), instance_id)
return InstancesManager.instances[key]
@staticmethod
def _get_session_id(session):
if not session:
def get_session_id(session):
if session is None:
return "** NOT LOGGED IN **"
if "user_info" not in session:
return "** UNKNOWN USER **"
return session["user_info"].get("id", "** INVALID SESSION **")
@staticmethod
def get_auth_proxy():
return InstancesManager.get(special_session, Ids.AuthProxy)
@staticmethod
def reset():
return InstancesManager.instances.clear()
InstancesManager.instances.clear()
RootInstance = SingleInstance(special_session, Ids.Root, None)
RootInstance = SingleInstance(None, special_session, Ids.Root)