# mypy: disable-error-code="override"
import logging
import os
from typing import Any
from pynamodb.attributes import UnicodeAttribute
from pynamodb.constants import PAY_PER_REQUEST_BILLING_MODE
from pynamodb.exceptions import DeleteError, DoesNotExist, PutError, UpdateError
from pynamodb.indexes import AllProjection, GlobalSecondaryIndex
from pynamodb.models import Model
from actingweb.db.dynamodb._ensure import ensure_table
from actingweb.db.exceptions import DbError
logger = logging.getLogger(__name__)
"""
DbProperty handles all db operations for a property
AWS DynamoDB is used as a backend.
"""
[docs]
class PropertyIndex(GlobalSecondaryIndex[Any]):
"""
Secondary index on property
"""
value = UnicodeAttribute(default="0", hash_key=True)
[docs]
class Property(Model):
"""
DynamoDB data model for a property.
Deliberately declares NO global secondary index: in lookup-table mode
(the reverse-lookup mechanism of record) the legacy value-keyed GSI
would only add write/storage amplification and DynamoDB's 2048-byte
GSI-key limit on every property value. Tables created through this
class therefore have no GSI. Legacy-mode deployments create the table
through :class:`PropertyLegacy` instead — the schema a deployment
creates matches the code path its configuration selects.
"""
id = UnicodeAttribute(hash_key=True)
name = UnicodeAttribute(range_key=True)
value = UnicodeAttribute()
[docs]
class PropertyLegacy(Model):
"""Legacy schema variant of the SAME properties table (deprecated).
Identical item shape to :class:`Property` plus the value-keyed
``property-index`` GSI. Used only (a) to create the table when the
deployment runs in legacy reverse-lookup mode, and (b) to query the
GSI on the legacy reverse-lookup path. All regular data-plane
operations go through :class:`Property` — the two classes are
item-compatible. Removed in the next major release together with the
legacy path.
"""
id = UnicodeAttribute(hash_key=True)
name = UnicodeAttribute(range_key=True)
value = UnicodeAttribute()
property_index = PropertyIndex()
def _serialize_property_value(value: Any) -> str | None:
"""Serialize a property value the same way ``DbProperty.set()`` does.
Returns ``None`` if the serialized value is empty (nothing to write —
callers treat this as "would delete", which conditional-create callers
must reject rather than silently no-op).
"""
import json
from actingweb.db.utils import sanitize_json_data
if value is not None and not isinstance(value, str):
try:
sanitized_value = sanitize_json_data(value, log_source="property")
value = json.dumps(sanitized_value)
except (TypeError, ValueError):
value = str(value)
elif isinstance(value, str):
value = sanitize_json_data(value, log_source="property")
if not value or (hasattr(value, "__len__") and len(value) == 0):
return None
return value
[docs]
class DbProperty:
"""
DbProperty does all the db operations for property objects
The actor_id must always be set. get(), set() and
get_actor_id_from_property() will set a new internal handle
that will be reused by set() (overwrite property) and
delete().
"""
def __init__(
self,
use_lookup_table: bool | None = None,
indexed_properties: list[str] | None = None,
) -> None:
"""Initialize DbProperty.
Args:
use_lookup_table: Whether to use property lookup table. If None, reads from env.
indexed_properties: List of property names to index. If None, uses defaults.
"""
self.handle: Property | PropertyLegacy | None = None
# Store configuration for lookup table (resolved before table
# creation — the mode decides which schema a fresh table gets)
if use_lookup_table is not None:
self._use_lookup_table = use_lookup_table
else:
self._use_lookup_table = (
os.getenv("USE_PROPERTY_LOOKUP_TABLE", "true").lower() == "true"
)
if indexed_properties is not None:
self._indexed_properties = indexed_properties
else:
self._indexed_properties = ["oauthId", "email", "externalUserId"]
if os.getenv("INDEXED_PROPERTIES"):
env_props = os.getenv("INDEXED_PROPERTIES", "").split(",")
self._indexed_properties = [p.strip() for p in env_props if p.strip()]
# Lookup mode creates the table WITHOUT the legacy GSI; legacy mode
# keeps it. Existing tables are never altered — first creator wins.
ensure_table(Property if self._use_lookup_table else PropertyLegacy)
def _should_index_property(self, name: str) -> bool:
"""
Check if property should be indexed in lookup table.
Returns True if:
1. Lookup table mode is enabled
2. Property name is in configured indexed_properties list
3. Property is not a list-property item/meta row (belt-and-braces —
list names are never configured as indexed properties, but this
makes it structurally impossible for lookup-table sync to touch
list storage rows)
"""
return (
self._use_lookup_table
and name in self._indexed_properties
and not name.startswith("list:")
)
[docs]
def get(self, actor_id: str | None = None, name: str | None = None) -> str | None:
"""Retrieves the property from the database.
Returns ``None`` only when the row is absent. A backend fault
(throttle, timeout, connection error) raises ``DbError`` instead of
being reported as absence.
"""
if not actor_id or not name:
return None
if self.handle is not None and (
str(self.handle.id) != actor_id or str(self.handle.name) != name
):
# A handle cached from a previous get()/set() call for a
# different (actor_id, name) must never be reused — discard it
# and take the fresh-fetch path below.
self.handle = None
if self.handle:
try:
self.handle.refresh()
except DoesNotExist:
return None
except Exception as e:
raise DbError("property read", actor_id) from e
return str(self.handle.value) if self.handle.value else None
try:
self.handle = Property.get(actor_id, name, consistent_read=True)
except DoesNotExist:
return None
except Exception as e:
raise DbError("property read", actor_id) from e
return str(self.handle.value) if self.handle.value else None
[docs]
def get_actor_id_from_property(
self, name: str | None = None, value: str | None = None
) -> str | None:
"""
Reverse lookup: find actor by property value.
Uses lookup table if configured, otherwise falls back to GSI.
Args:
name: Property name (e.g., "oauthId")
value: Property value to search for
Returns:
Actor ID if found, None otherwise
"""
if not name or not value:
return None
if self._use_lookup_table:
if name not in self._indexed_properties:
# Enforce the documented contract: only properties configured
# via with_indexed_properties() support reverse lookup. The
# old behaviour silently fell through to the legacy GSI,
# which crashes on tables created without that index.
logger.warning(
f"Reverse lookup requested for non-indexed property "
f"'{name}' — add it to with_indexed_properties() (or "
f"INDEXED_PROPERTIES) to enable reverse lookup; "
f"returning None"
)
return None
# Use lookup table approach
from actingweb.db.dynamodb.property_lookup import DbPropertyLookup
lookup = DbPropertyLookup()
actor_id = lookup.get(property_name=name, value=value)
if actor_id is None:
# Migration fallback tier 1: the deprecated v1 lookup table
# (deployments that adopted lookup mode before the v2 digest
# format). A hit means the v2 backfill has not run yet.
actor_id = lookup.get_v1(property_name=name, value=value)
if actor_id:
logger.warning(
f"DEPRECATED: reverse lookup for '{name}' served from "
f"the v1 lookup table — run "
f"scripts/backfill_property_lookup.py to migrate to "
f"the v2 format, then drop the v1 table. This "
f"fallback is removed in the next major release."
)
if actor_id is None:
# Migration fallback tier 2: the legacy value-keyed GSI
# (deployments upgrading from legacy mode with an un-backfilled
# lookup table). Missing index/table is a normal state.
try:
# The GSI is keyed on value alone; filter by name so a
# same-value/different-property row on another actor
# can't hijack the lookup.
for res in PropertyLegacy.property_index.query(
value, filter_condition=PropertyLegacy.name == name
):
actor_id = str(res.id) if res.id else None
break
except Exception:
actor_id = None
if actor_id:
logger.warning(
f"DEPRECATED: reverse lookup for '{name}' served from "
f"the legacy property-index GSI — run "
f"scripts/backfill_property_lookup.py to populate the "
f"lookup table. This fallback is removed in the next "
f"major release."
)
if actor_id:
# Load the property into self.handle for subsequent operations
try:
self.handle = Property.get(actor_id, name, consistent_read=True)
except Exception:
logger.warning(
f"Lookup found actor {actor_id} but property {name} doesn't exist"
)
return None
return actor_id
else:
# Legacy GSI approach (deprecated)
try:
results = PropertyLegacy.property_index.query(value)
self.handle = None
for res in results:
self.handle = res
break
except Exception as e:
if "index" in str(e).lower() or "ValidationException" in str(e):
raise RuntimeError(
f"Legacy property-index GSI is missing from table "
f"'{Property.Meta.table_name}'. Reverse lookup is "
f"configured to use the legacy DynamoDB GSI "
f"(use_lookup_table=False), but this table has no "
f"'property-index' GSI — it was created without one, "
f"and the library cannot add a GSI to a live table. "
f"Three ways to resolve: "
f"(1) RECOMMENDED: switch to lookup-table mode with "
f"with_legacy_property_index(enable=False) and run "
f"scripts/backfill_property_lookup.py; "
f"(2) keep legacy mode and add the GSI to the live "
f"table via 'aws dynamodb update-table' — note this "
f"imposes a 2048-byte limit on ALL property values; "
f"(3) disable reverse lookup entirely with "
f"with_indexed_properties([]). "
f"See docs/migration/v3.13 for details."
) from e
raise
if not self.handle:
return None
return str(self.handle.id) if self.handle.id else None
[docs]
def set(
self, actor_id: str | None = None, name: str | None = None, value: Any = None
) -> bool:
"""Sets a new value for the property name"""
if not name:
return False
# Convert non-string values to JSON strings for storage
import json
from actingweb.db.utils import sanitize_json_data
if value is not None and not isinstance(value, str):
try:
# Defensive sanitization of own data before JSON encoding
sanitized_value = sanitize_json_data(value, log_source="property")
value = json.dumps(sanitized_value)
except (TypeError, ValueError):
value = str(value)
elif isinstance(value, str):
# Sanitize string values too — surrogates in pre-serialized JSON
# strings bypass json.dumps sanitization and corrupt storage
value = sanitize_json_data(value, log_source="property")
# Handle empty value (deletion)
if not value or (hasattr(value, "__len__") and len(value) == 0):
if self.get(actor_id=actor_id, name=name):
self.delete() # This will also delete lookup entry
return True
if (
actor_id
and self.handle is not None
and (str(self.handle.id) != actor_id or str(self.handle.name) != name)
):
# Same rule as get(): a handle cached for a different
# (actor_id, name) must never be reused.
self.handle = None
# Get old value before updating (for lookup sync)
old_value = None
if self._should_index_property(name):
if self.handle and self.handle.value:
old_value = str(self.handle.value)
elif actor_id:
# PropertyStore.__setattr__ builds a fresh DbProperty for
# every write, so self.handle is unset on the primary public
# path. Read the current stored value (like the PostgreSQL
# backend) so changing an indexed value deletes its stale
# lookup row instead of leaving it to resolve forever.
old_value = self.get(actor_id=actor_id, name=name)
# Save property
if not self.handle:
if not actor_id:
return False
self.handle = Property(id=actor_id, name=name, value=value)
else:
self.handle.value = value
try:
self.handle.save()
except Exception as e:
raise DbError("property write", actor_id) from e
# Update lookup table if property is indexed
if self._should_index_property(name):
# Use handle.id which is guaranteed to be set after save()
handle_actor_id = str(self.handle.id) if self.handle.id else actor_id
if handle_actor_id:
self._update_lookup_entry(handle_actor_id, name, old_value, value)
return True
def _update_lookup_entry(
self, actor_id: str, name: str, old_value: str | None, new_value: str
) -> None:
"""
Update lookup table entry (delete old, create new).
Best-effort update - logs errors but doesn't fail property write.
"""
# Unchanged value: nothing to sync — avoids a delete+put per
# repeated write of the same indexed value.
if old_value is not None and old_value == new_value:
return
try:
from actingweb.db.dynamodb.property_lookup import DbPropertyLookup
db = DbPropertyLookup()
# Delete the old entry if it exists and belongs to this actor.
# Note: theoretical race if another actor creates the same value
# between get() and delete(); best-effort design accepts this.
if old_value:
if db.get(property_name=name, value=old_value) == actor_id:
db.delete()
# Conditional create: a row owned by another actor is a logged
# collision, not a silent overwrite (see DbPropertyLookup.create).
db.create(property_name=name, value=new_value, actor_id=actor_id)
except Exception as e:
logger.error(
f"LOOKUP_TABLE_SYNC_FAILED: actor={actor_id} property={name} "
f"old_value_len={len(old_value) if old_value else 0} "
f"new_value_len={len(new_value)} error={e}"
)
# Don't fail the property write - accept eventual consistency
def _delete_lookup_entry(self, actor_id: str | None, name: str, value: str) -> None:
"""
Delete lookup table entry.
Best-effort deletion - logs errors but doesn't fail property delete.
"""
try:
from actingweb.db.dynamodb.property_lookup import DbPropertyLookup
db = DbPropertyLookup()
# Verify it belongs to the same actor before deleting
if db.get(property_name=name, value=value) == actor_id:
db.delete()
except Exception as e:
logger.warning(
f"LOOKUP_DELETE_FAILED: actor={actor_id} property={name} "
f"value_len={len(value)} error={e}"
)
# Don't fail the property delete
[docs]
def delete(self) -> bool:
"""Deletes the property in the database after a get()"""
if not self.handle:
return False
# Save values before deletion
actor_id = str(self.handle.id) if self.handle.id else None
name = str(self.handle.name) if self.handle.name else None
value = str(self.handle.value) if self.handle.value else None
# Delete property
self.handle.delete()
self.handle = None
# Delete lookup entry if property is indexed
if name and value and self._should_index_property(name):
self._delete_lookup_entry(actor_id, name, value)
return True
[docs]
def get_range(
self,
actor_id: str | None = None,
lower: str | None = None,
upper: str | None = None,
keys_only: bool = False,
consistent_read: bool = True,
) -> dict[str, str]:
"""Range-read rows whose name is in ``[lower, upper]``.
See ``DbPropertyProtocol.get_range`` for the contract. DynamoDB's
KeyConditionExpression rejects two separate comparisons on the same
key (``>=`` AND ``<`` is invalid), so this uses ``between()``,
which is INCLUSIVE on both ends — the caller MUST choose ``upper``
as a sentinel value that can never equal a real row name (e.g. a
delimiter character no real key contains), so inclusive-vs-exclusive
at the boundary is unobservable. DynamoDB already returns range-key
query results in ascending sort-key order, but this is NOT relied
upon — the caller re-sorts.
"""
if not actor_id or lower is None or upper is None:
return {}
condition = Property.name.between(lower, upper)
attributes_to_get = ["name"] if keys_only else ["name", "value"]
try:
results: dict[str, str] = {}
for item in Property.query(
actor_id,
range_key_condition=condition,
consistent_read=consistent_read,
attributes_to_get=attributes_to_get,
):
results[str(item.name)] = "" if keys_only else str(item.value or "")
return results
except Exception as e:
raise DbError("property range read", actor_id) from e
[docs]
def get_prefix(
self,
actor_id: str | None = None,
prefix: str | None = None,
keys_only: bool = False,
consistent_read: bool = True,
) -> dict[str, str]:
"""Read rows whose name begins with ``prefix``.
See ``DbPropertyProtocol.get_prefix`` for the contract. Uses
DynamoDB's native ``begins_with`` on the range key, which is EXACT
for an arbitrary UTF-8 prefix: String sort keys are ordered by
their UTF-8 bytes, and UTF-8 is prefix-preserving, so "sorts under
this prefix" and "starts with these bytes" are the same set. That
is why this is not a ``get_range`` with a synthesised upper bound —
no such bound is exact.
``begins_with`` performs NO Unicode normalization, so an NFD prefix
does not match an NFC name. This is deliberate and matches
PostgreSQL's ``starts_with()``; it is what makes the two backends
return byte-identical key sets.
The empty prefix is rejected here rather than passed down:
``begins_with(name, "")`` is a ``ValidationException``.
"""
if not actor_id or not prefix:
return {}
condition = Property.name.startswith(prefix)
attributes_to_get = ["name"] if keys_only else ["name", "value"]
try:
results: dict[str, str] = {}
for item in Property.query(
actor_id,
range_key_condition=condition,
consistent_read=consistent_read,
attributes_to_get=attributes_to_get,
):
results[str(item.name)] = "" if keys_only else str(item.value or "")
return results
except Exception as e:
raise DbError("property prefix read", actor_id) from e
[docs]
def create_if_not_exists(
self, actor_id: str | None = None, name: str | None = None, value: Any = None
) -> bool:
"""Conditionally create a row — see ``DbPropertyProtocol.create_if_not_exists``."""
if not actor_id or not name:
return False
serialized = _serialize_property_value(value)
if serialized is None:
return False
item = Property(id=actor_id, name=name, value=serialized)
try:
item.save(condition=Property.id.does_not_exist())
except PutError as e:
if e.cause_response_code == "ConditionalCheckFailedException":
return False
raise DbError("property conditional create", actor_id) from e
except Exception as e:
raise DbError("property conditional create", actor_id) from e
return True
[docs]
def delete_if_value_equals(
self, actor_id: str | None = None, name: str | None = None, value: Any = None
) -> bool:
"""Conditionally delete — see ``DbPropertyProtocol.delete_if_value_equals``.
The condition covers both "someone changed it" and "someone already
deleted it": DynamoDB fails an equality condition on a missing
attribute just as it does on a differing one, and both mean the same
thing to the caller (re-resolve and retry).
"""
if not actor_id or not name or value is None:
return False
item = Property(id=actor_id, name=name)
try:
item.delete(condition=Property.value == value)
except DeleteError as e:
if e.cause_response_code == "ConditionalCheckFailedException":
return False
raise DbError("property conditional delete", actor_id) from e
except Exception as e:
raise DbError("property conditional delete", actor_id) from e
return True
[docs]
def set_if_value_equals(
self,
actor_id: str | None = None,
name: str | None = None,
expected: Any = None,
value: Any = None,
) -> bool:
"""Conditionally set — see ``DbPropertyProtocol.set_if_value_equals``.
Same condition-failure-vs-fault distinction as
``delete_if_value_equals``: a differing stored value and a vanished
row both fail the equality condition on ``value`` the same way.
"""
if not actor_id or not name or expected is None or value is None:
return False
item = Property(id=actor_id, name=name)
try:
item.update(
actions=[Property.value.set(value)],
condition=Property.value == expected,
)
except UpdateError as e:
if e.cause_response_code == "ConditionalCheckFailedException":
return False
raise DbError("property conditional set", actor_id) from e
except Exception as e:
raise DbError("property conditional set", actor_id) from e
return True
[docs]
def get_last_in_range(
self,
actor_id: str | None = None,
lower: str | None = None,
upper: str | None = None,
) -> str | None:
"""Bytewise-greatest row name in ``[lower, upper]`` — see
``DbPropertyProtocol.get_last_in_range``.
``scan_index_forward=False, limit=1`` reads DynamoDB's natural
range-key sort order backwards and stops at the first item — one
item's read capacity, not the whole range's.
"""
if not actor_id or lower is None or upper is None:
return None
condition = Property.name.between(lower, upper)
try:
for item in Property.query(
actor_id,
range_key_condition=condition,
consistent_read=True,
scan_index_forward=False,
limit=1,
attributes_to_get=["name"],
):
return str(item.name)
return None
except Exception as e:
raise DbError("property last-in-range read", actor_id) from e
[docs]
def batch_delete(
self, actor_id: str | None = None, names: list[str] | None = None
) -> None:
"""Unconditional bulk delete — see ``DbPropertyProtocol.batch_delete``.
``Property.batch_write()`` is PynamoDB's ``BatchWrite`` context
manager: it chunks at 25 items (``BATCH_WRITE_PAGE_LIMIT``) and
retries any items DynamoDB reports as unprocessed, raising
``PutError`` if ``Meta.max_retry_attempts`` is exhausted -- both
behaviours come from the pinned PynamoDB version, not hand-rolled
here.
"""
if not actor_id or not names:
return
try:
with Property.batch_write() as batch:
for name in names:
batch.delete(Property(id=actor_id, name=name))
except Exception as e:
raise DbError("property batch delete", actor_id) from e
[docs]
class DbPropertyList:
"""
DbPropertyList does all the db operations for list of property objects
The actor_id must always be set.
"""
def __init__(
self,
use_lookup_table: bool | None = None,
indexed_properties: list[str] | None = None,
) -> None:
"""Initialize DbPropertyList.
Args:
use_lookup_table: Whether to use property lookup table. If None, reads from env.
indexed_properties: List of property names to index. If None, uses defaults.
"""
self.handle: Any | None = None
self.actor_id: str | None = None
self.props: dict[str, str] | None = None
if use_lookup_table is not None:
self._use_lookup_table = use_lookup_table
else:
self._use_lookup_table = (
os.getenv("USE_PROPERTY_LOOKUP_TABLE", "true").lower() == "true"
)
if indexed_properties is not None:
self._indexed_properties = indexed_properties
else:
self._indexed_properties = ["oauthId", "email", "externalUserId"]
if os.getenv("INDEXED_PROPERTIES"):
env_props = os.getenv("INDEXED_PROPERTIES", "").split(",")
self._indexed_properties = [p.strip() for p in env_props if p.strip()]
# Lookup mode creates the table WITHOUT the legacy GSI; legacy mode
# keeps it. Existing tables are never altered — first creator wins.
ensure_table(Property if self._use_lookup_table else PropertyLegacy)
[docs]
def fetch(self, actor_id: str | None = None) -> dict[str, str] | None:
"""Retrieves the PLAIN (non-list) properties of an actor_id from
the database.
Two range-constrained Queries rather than a whole-partition Query
with client-side filtering: DynamoDB cannot ``OR`` on a sort key,
so excluding the ``list:``-prefixed rows takes a pair of Queries
covering everything below and everything above that namespace,
instead of paying for (and discarding) every list item row on
every plain-property read.
The upper sentinel is ``"list;"`` (``;`` is 0x3B, the byte right
after ``:``), NOT ``"list:~"``: ``~`` is 0x7E, so any list whose
NAME starts with a byte above ``~`` -- every non-ASCII list name --
would sort after ``"list:~"`` and leak back into this result.
``name < "list:"`` and ``name >= "list;"`` is exact: a plain
property named ``"list"`` sorts in the first range, one named
``"listen"`` in the second, and every ``list:*`` row in neither.
"""
if not actor_id:
return None
self.actor_id = actor_id
self.props = {}
# Property.query() returns a truthy iterator regardless of match
# count, so fetch() has always returned {} (not None) for an actor
# whose partition holds no plain properties -- preserved here by
# unconditionally returning self.props rather than reintroducing a
# falsy-handle check against either query's result.
for condition in (Property.name < "list:", Property.name >= "list;"):
self.handle = Property.query(actor_id, range_key_condition=condition)
for d in self.handle:
self.props[d.name] = d.value
return self.props
[docs]
def fetch_all_including_lists(
self, actor_id: str | None = None
) -> dict[str, str] | None:
"""Retrieves ALL properties including list properties - for internal PropertyListStore use"""
if not actor_id:
return None
self.actor_id = actor_id
self.handle = Property.query(actor_id)
if self.handle:
props = {}
for d in self.handle:
props[d.name] = d.value
return props
else:
return None
[docs]
def delete(self) -> bool:
"""Deletes all the properties in the database"""
if not self.actor_id:
return False
# Single partition read: collect indexed properties (for lookup-row
# cleanup) and item names, then delete everything in one batched
# operation (Phase 12: thoughts/plans/2026-08-20-v2-positional-
# access-cost.md) instead of one DeleteItem round trip per row.
indexed_props: list[tuple[str, str]] = []
names: list[str] = []
self.handle = Property.query(self.actor_id)
for p in self.handle:
if self._use_lookup_table and str(p.name) in self._indexed_properties:
indexed_props.append((str(p.name), str(p.value)))
names.append(str(p.name))
if names:
try:
with Property.batch_write() as batch:
for name in names:
batch.delete(Property(id=self.actor_id, name=name))
except Exception as e:
raise DbError("property batch delete", self.actor_id) from e
# Delete lookup entries
if indexed_props:
from actingweb.db.dynamodb.property_lookup import DbPropertyLookup
db = DbPropertyLookup()
for name, value in indexed_props:
try:
# Verify ownership before deleting: a shared indexed value
# may map to another actor's row, which must survive this
# actor's deletion.
if db.get(property_name=name, value=value) == self.actor_id:
db.delete()
except Exception as e:
logger.warning(
f"Failed to delete lookup entry for property {name}: {e}"
)
self.handle = None
return True