Source code for actingweb.db.dynamodb.attribute

import os
import time
from collections.abc import Sequence

from pynamodb.attributes import (
    JSONAttribute,
    NumberAttribute,
    UnicodeAttribute,
    UTCDateTimeAttribute,
)
from pynamodb.constants import PAY_PER_REQUEST_BILLING_MODE
from pynamodb.exceptions import DoesNotExist
from pynamodb.models import Model

from actingweb.db.dynamodb._ensure import ensure_table

"""
    DbAttribute handles all db operations for an attribute (internal)
    AWS DynamoDB is used as a backend.
"""


[docs] class Attribute(Model): """ DynamoDB data model for a property """
[docs] class Meta: # type: ignore[misc] table_name = os.getenv("AWS_DB_PREFIX", "demo_actingweb") + "_attributes" billing_mode = PAY_PER_REQUEST_BILLING_MODE region = os.getenv("AWS_DEFAULT_REGION", "us-west-1") host = os.getenv("AWS_DB_HOST", None)
id = UnicodeAttribute(hash_key=True) bucket_name = UnicodeAttribute(range_key=True) bucket = UnicodeAttribute() name = UnicodeAttribute() data = JSONAttribute(null=True) timestamp = UTCDateTimeAttribute(null=True) # TTL timestamp for automatic DynamoDB expiration (Unix epoch timestamp) # Enable DynamoDB TTL on this field for automatic cleanup ttl_timestamp = NumberAttribute(null=True)
[docs] class DbAttribute: """ DbProperty does all the db operations for property objects The actor_id must always be set. get(), set() will set a new internal handle that will be reused by set() (overwrite attribute) and delete(). """
[docs] @staticmethod def get_bucket(actor_id=None, bucket=None): """Returns a dict of attributes from a bucket, each with data and timestamp ``bucket_name`` is ``bucket + ":" + name``, so ``begins_with(bucket)`` would also match every bucket that has this one as a prefix — bucket ``remote:abc`` seeing ``remote:abcd``'s rows. The delimiter narrows the Query, and the exact ``t.bucket`` compare is what actually separates them, because both halves of the composite key may themselves contain ``:`` (bucket ``remote:{peer_id}``, name ``list:{name}:{index}``), so bucket ``remote:abc``/name ``x`` and bucket ``remote``/name ``abc:x`` produce an identical ``bucket_name``. Same guard as ``delete_by_chain()`` below and ``db/dynamodb/subscription_suspension.py``'s cascade check. """ if not actor_id or not bucket: return None try: query = Attribute.query( actor_id, Attribute.bucket_name.startswith(bucket + ":"), consistent_read=True, ) except Exception: # PynamoDB DoesNotExist exception return None ret = {} for t in query: if t.bucket != bucket: continue ret[t.name] = { "data": t.data, "timestamp": t.timestamp, } return ret
[docs] @staticmethod def get_attr(actor_id=None, bucket=None, name=None): """Returns a dict of attributes from a bucket, each with data and timestamp""" if not actor_id or not bucket or not name: return None try: r = Attribute.get(actor_id, bucket + ":" + name, consistent_read=True) except Exception: # PynamoDB DoesNotExist exception return None # Same guard as get_bucket(): the composite key can collide across # buckets, and the stored ``bucket`` is what says whose row it is. if r.bucket != bucket: return None return { "data": r.data, "timestamp": r.timestamp, }
[docs] @staticmethod def get_attr_strict(actor_id=None, bucket=None, name=None): """Point-read one attribute, distinguishing absence from failure. ``get_attr()`` collapses every outcome into ``None``: a missing row, a throttle, an expired credential and a missing table are indistinguishable. That is tolerable for a cache read and wrong for any caller whose "row is absent" branch is a decision rather than a fallback — the deletion tombstone read being the case in the library. Absence returns ``None``; anything else raises so the caller can tell "I looked and it is not there" from "I could not look". An expired row reads as absent. DynamoDB's TTL sweep may lag by up to 48 hours, and a tombstone past its window must stop suppressing writes on time rather than whenever the sweeper gets to it. """ if not actor_id or not bucket or not name: return None try: r = Attribute.get(actor_id, bucket + ":" + name, consistent_read=True) except DoesNotExist: return None if r.bucket != bucket: return None if r.ttl_timestamp is not None and r.ttl_timestamp <= int(time.time()): return None return { "data": r.data, "timestamp": r.timestamp, }
[docs] @staticmethod def set_attr( actor_id=None, bucket=None, name=None, data=None, timestamp=None, ttl_seconds=None, ): """Sets a data value for a given attribute in a bucket. Args: actor_id: The actor ID bucket: The bucket name name: The attribute name data: The data to store (JSON-serializable) timestamp: Optional timestamp ttl_seconds: Optional TTL in seconds from now. If provided, DynamoDB will automatically delete this item after expiry. Note: A 1-hour buffer is added for clock skew safety. """ if not actor_id or not name or not bucket: return False if not data: try: item = Attribute.get( actor_id, bucket + ":" + name, consistent_read=True ) if item.bucket == bucket: item.delete() except Exception: # PynamoDB DoesNotExist exception pass return True # Calculate TTL timestamp if provided ttl_timestamp = None if ttl_seconds is not None: from ...constants import TTL_CLOCK_SKEW_BUFFER # Add buffer for clock skew safety ttl_timestamp = int(time.time()) + ttl_seconds + TTL_CLOCK_SKEW_BUFFER # Defensive sanitization of data before storing from actingweb.db.utils import sanitize_json_data data = sanitize_json_data(data, log_source="attribute") new = Attribute( id=actor_id, bucket_name=bucket + ":" + name, bucket=bucket, name=name, data=data, timestamp=timestamp, ttl_timestamp=ttl_timestamp, ) new.save() return True
[docs] def delete_attr(self, actor_id=None, bucket=None, name=None): """Deletes an attribute in a bucket""" return self.set_attr(actor_id=actor_id, bucket=bucket, name=name, data=None)
[docs] @staticmethod def delete_attr_conditional(actor_id=None, bucket=None, name=None): """Atomically delete an attribute, returning True only if THIS call removed an existing item. The DeleteItem carries a ``attribute_exists(id)`` condition, so when two callers race on the same attribute exactly one delete succeeds and the other fails its condition check (item already gone). Backs single-use/atomic-consume semantics (e.g. mobile-ticket redemption). Args: actor_id: The actor ID bucket: The bucket name name: The attribute name Returns: True if this call removed an existing item, False otherwise """ if not actor_id or not bucket or not name: return False try: item = Attribute.get(actor_id, bucket + ":" + name, consistent_read=True) except Exception: # PynamoDB DoesNotExist exception return False if item.bucket != bucket: return False try: # Condition fails (raises) if a concurrent caller already deleted it. item.delete(condition=Attribute.id.exists()) return True except Exception: return False
[docs] @staticmethod def conditional_update_attr( actor_id=None, bucket=None, name=None, old_data=None, new_data=None, timestamp=None, ): # type: ignore[misc] """Conditionally update an attribute only if current data matches old_data. This provides atomic compare-and-swap functionality for race-free updates. JSON comparison is order-independent - dict key ordering does not affect equality. If the caller's old_data has different key ordering than stored data, we normalize both sides for comparison and use the stored ordering for the atomic update. Args: actor_id: The actor ID bucket: The bucket name name: The 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 """ import json if not actor_id or not bucket or not name: return False bucket_name = bucket + ":" + name # Defensive sanitization of data before storing from actingweb.db.utils import sanitize_json_data old_data = sanitize_json_data(old_data, log_source="attribute") new_data = sanitize_json_data(new_data, log_source="attribute") try: # Get current item with consistent read item = Attribute.get(actor_id, bucket_name, consistent_read=True) if item.bucket != bucket: return False # Normalize JSON for order-independent comparison # This ensures {"a": 1, "b": 2} == {"b": 2, "a": 1} def normalize_json(data): """Normalize JSON data by serializing with sorted keys.""" if data is None: return None return json.loads(json.dumps(data, sort_keys=True)) old_data_normalized = normalize_json(old_data) current_data_normalized = normalize_json(item.data) # Check if current data matches old_data (order-independent) if old_data_normalized != current_data_normalized: return False # Data matches semantically - perform atomic update # Use the ACTUAL stored data for DynamoDB's condition to ensure atomicity # This handles the case where old_data has different key ordering actions: Sequence[object] = [Attribute.data.set(new_data)] if timestamp: actions = list(actions) + [Attribute.timestamp.set(timestamp)] item.update( actions=actions, # type: ignore[arg-type] condition=( Attribute.data == item.data ), # Atomic check against current value ) return True except Exception: # Item doesn't exist or condition check failed (race condition) return False
[docs] @staticmethod def delete_bucket(actor_id=None, bucket=None): """Deletes an entire bucket Carries ``get_bucket()``'s guard, and needs it more: without the delimiter and the exact ``t.bucket`` compare this deletes the rows of every bucket that has this one as a prefix. ``RemotePeerStore`` deletes bucket ``remote:{peer_id}`` on trust teardown and most call sites build that id with ``validate_peer_id=False``, so ending trust with peer ``abc`` would destroy peer ``abcd``'s dataset. Same guard as ``delete_by_chain()`` below. """ if not actor_id or not bucket: return False try: query = Attribute.query( actor_id, Attribute.bucket_name.startswith(bucket + ":"), consistent_read=True, ) except Exception: # PynamoDB DoesNotExist exception return True for t in query: if t.bucket != bucket: continue t.delete() return True
[docs] @staticmethod def delete_expired(now_epoch=None, buckets=None): # type: ignore[misc] """Purge TTL-expired attributes. DynamoDB deletes expired items automatically when TTL is enabled on the ``ttl_timestamp`` attribute of the table (recommended; the field is written with a clock-skew buffer for exactly this purpose). A backend-driven purge here would require a full-table ``Scan``, which is intentionally avoided, so this relies on native TTL and reports 0 deletions. Ensure DynamoDB TTL is enabled on the attributes table. Args: now_epoch: Ignored (native TTL uses the stored ttl_timestamp). buckets: Ignored. Returns: 0 — deletion is handled asynchronously by DynamoDB TTL. """ return 0
[docs] @staticmethod def delete_by_chain(actor_id=None, buckets=None, chain_id=None): # type: ignore[misc] """Delete attributes whose stored ``data['chain_id']`` matches chain_id. Backs refresh-token family (chain) revocation. DynamoDB has no secondary index on the JSON-embedded ``chain_id``, so this queries the (shared) token buckets and filters in memory. The cost is bounded by the shortened used-token TTL that keeps the buckets small; for very large deployments the optimization path is a GSI on a promoted top-level ``chain_id`` attribute. Revocation is rare (a theft event), so the scan is acceptable. Args: actor_id: Storage partition id (the system actor the tokens live under). buckets: Bucket whitelist (the SPA access + refresh token buckets). chain_id: The refresh-token family identifier to delete. Returns: Number of items deleted. """ if not actor_id or not chain_id or not buckets: return 0 deleted = 0 for bucket in buckets: try: query = Attribute.query( actor_id, Attribute.bucket_name.startswith(bucket), consistent_read=True, ) except Exception: # PynamoDB DoesNotExist exception continue for t in list(query): data = t.data if ( t.bucket == bucket and isinstance(data, dict) and data.get("chain_id") == chain_id ): t.delete() deleted += 1 return deleted
def __init__(self): ensure_table(Attribute)
[docs] class DbAttributeBucketList: """ DbAttributeBucketList handles multiple buckets The actor_id must always be set. """
[docs] @staticmethod def fetch(actor_id=None): """Retrieves all the attributes of an actor_id from the database""" if not actor_id: return None try: query = Attribute.query(actor_id) except Exception: # PynamoDB DoesNotExist exception return None ret = {} for t in query: if t.bucket not in ret: ret[t.bucket] = {} ret[t.bucket][t.name] = { "data": t.data, "timestamp": t.timestamp, } return ret
[docs] @staticmethod def fetch_timestamps(actor_id=None): """Retrieves timestamps for all buckets of an actor_id""" if not actor_id: return None try: query = Attribute.query(actor_id) except Exception: # PynamoDB DoesNotExist exception return None ret = {} for t in query: if t.bucket not in ret: ret[t.bucket] = t.timestamp return ret
[docs] @staticmethod def delete(actor_id=None): """Deletes all the attributes in the database""" if not actor_id: return False try: query = Attribute.query(actor_id) except Exception: # PynamoDB DoesNotExist exception return False for t in query: t.delete() return True
def __init__(self): ensure_table(Attribute)