from typing import Any, cast
from actingweb.db import get_attribute, get_attribute_bucket_list
[docs]
class InternalStore:
"""Access to internal attributes using .prop notation"""
def __init__(
self,
actor_id: str | None = None,
config: Any | None = None,
bucket: str | None = None,
) -> None:
if not bucket:
bucket = "_internal"
self._db = Attributes(actor_id=actor_id, bucket=bucket, config=config)
d = self._db.get_bucket()
if d:
for k, v in d.items():
self.__setattr__(k, v.get("data"))
self.__initialised = True
def __getitem__(self, k: str) -> Any:
return self.__getattr__(k)
def __setitem__(self, k: str, v: Any) -> None:
return self.__setattr__(k, v)
def __setattr__(self, k: str, v: Any) -> None:
if "_InternalStore__initialised" not in self.__dict__:
return object.__setattr__(self, k, v)
if k is None:
raise ValueError
if v is None:
self.__dict__["_db"].delete_attr(name=k)
if k in self.__dict__:
self.__delattr__(k)
else:
self.__dict__[k] = v
self.__dict__["_db"].set_attr(name=k, data=v)
def __getattr__(self, k: str) -> Any:
try:
return self.__dict__[k]
except KeyError:
return None
[docs]
class Attributes:
"""
Attributes is the main entity keeping an attribute.
It needs to be initalized at object creation time.
"""
[docs]
def get_bucket(self) -> dict[str, Any] | None:
"""Retrieves the attribute bucket from the database"""
if not self.data or len(self.data) == 0:
if self.dbprop:
fetched_data = self.dbprop.get_bucket(
actor_id=self.actor_id, bucket=self.bucket
)
# PostgreSQL backend returns None for non-existent buckets
if fetched_data is None:
self.data = {}
else:
# Cast needed due to dict invariance in value types
self.data = cast(dict[str, dict[str, Any] | None], fetched_data)
else:
self.data = {}
return self.data
[docs]
def get_attr(self, name: str | None = None) -> dict[str, Any] | None:
"""Retrieves a single attribute"""
if not name:
return None
# Ensure self.data is initialized (defensive check)
if self.data is None:
self.data = {}
if name not in self.data:
if self.dbprop:
self.data[name] = self.dbprop.get_attr(
actor_id=self.actor_id, bucket=self.bucket, name=name
)
else:
self.data[name] = None
return self.data[name]
[docs]
def set_attr(
self,
name: str | None = None,
data: Any | None = None,
timestamp: Any | None = None,
ttl_seconds: int | None = None,
) -> bool:
"""Sets new data for this attribute.
Args:
name: Attribute name
data: Data to store (JSON-serializable)
timestamp: Optional timestamp
ttl_seconds: Optional TTL in seconds. If provided, DynamoDB will
automatically delete this item after expiry.
"""
if not self.actor_id or not self.bucket or not name:
return False
# Ensure self.data is initialized (defensive check)
if self.data is None:
self.data = {}
assert self.data is not None # Type narrowing for pyright
if name not in self.data or self.data[name] is None:
self.data[name] = {}
attr_data = self.data[name]
assert attr_data is not None # Type narrowing for pyright
attr_data["data"] = data
attr_data["timestamp"] = timestamp
if self.dbprop:
return self.dbprop.set_attr(
actor_id=self.actor_id,
bucket=self.bucket,
name=name,
data=data,
timestamp=timestamp,
ttl_seconds=ttl_seconds,
)
return False
[docs]
def conditional_update_attr(
self,
name: str | None = None,
old_data: Any | None = None,
new_data: Any | None = None,
timestamp: Any | None = None,
) -> bool:
"""Conditionally update an attribute only if current data matches old_data.
This provides atomic compare-and-swap functionality for race-free updates.
Args:
name: Attribute name
old_data: Expected current data value (for comparison)
new_data: New data to set if current matches old_data
timestamp: Optional timestamp
Returns:
True if update succeeded (current matched old_data), False otherwise
"""
if not self.actor_id or not self.bucket or not name:
return False
if not self.dbprop:
return False
# Use the database backend's atomic conditional update
success = self.dbprop.conditional_update_attr(
actor_id=self.actor_id,
bucket=self.bucket,
name=name,
old_data=old_data,
new_data=new_data,
timestamp=timestamp,
)
# Update local cache only if successful
if success:
if self.data is None:
self.data = {}
assert self.data is not None # Type narrowing for pyright
if name not in self.data or self.data[name] is None:
self.data[name] = {}
attr_data = self.data[name]
assert attr_data is not None # Type narrowing for pyright
attr_data["data"] = new_data
attr_data["timestamp"] = timestamp
return success
[docs]
def delete_attr(self, name: str | None = None) -> bool:
if not name:
return False
if self.data and name in self.data:
del self.data[name]
if self.dbprop:
return self.dbprop.delete_attr(
actor_id=self.actor_id, bucket=self.bucket, name=name
)
return False
[docs]
def delete_bucket(self) -> bool:
"""Deletes the attribute bucket in the database"""
if not self.dbprop:
return False
if self.dbprop.delete_bucket(actor_id=self.actor_id, bucket=self.bucket):
if self.config:
self.dbprop = get_attribute(self.config)
else:
self.dbprop = None
self.data = {}
return True
else:
return False
def __init__(
self,
actor_id: str | None = None,
bucket: str | None = None,
config: Any | None = None,
) -> None:
"""A attribute must be initialised with actor_id and bucket"""
self.config = config
if self.config:
self.dbprop = get_attribute(self.config)
else:
self.dbprop = None
self.bucket = bucket
self.actor_id = actor_id
self.data: dict[str, dict[str, Any] | None] = {}
if actor_id and bucket and len(bucket) > 0 and config:
self.get_bucket()
[docs]
class Buckets:
"""Handles all attribute buckets of a specific actor_id
Access the attributes
in .props as a dictionary
"""
[docs]
def fetch(self) -> dict[str, dict[str, dict[str, Any]]] | bool:
if not self.actor_id:
return False
if self.list:
result = self.list.fetch(actor_id=self.actor_id)
return result if result is not None else False
return False
[docs]
def fetch_timestamps(self) -> dict[str, Any] | bool:
if not self.actor_id:
return False
if self.list:
result = self.list.fetch_timestamps(actor_id=self.actor_id)
return result if result is not None else False
return False
[docs]
def delete(self) -> bool:
if not self.list:
return False
self.list.delete(actor_id=self.actor_id)
if self.config:
self.list = get_attribute_bucket_list(self.config)
else:
self.list = None
return True
def __init__(self, actor_id: str | None = None, config: Any | None = None) -> None:
"""attributes must always be initialised with an actor_id"""
self.config = config
if not actor_id:
self.list = None
return
if self.config:
self.list = get_attribute_bucket_list(self.config)
else:
self.list = None
self.actor_id = actor_id