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 DoesNotExist
from pynamodb.indexes import AllProjection, GlobalSecondaryIndex
from pynamodb.models import Model
from actingweb.db.dynamodb._ensure import ensure_table
"""
DbActor handles all db operations for an actor
Google datastore for google is used as a backend.
"""
logger = logging.getLogger(__name__)
[docs]
class CreatorIndex(GlobalSecondaryIndex[Any]):
"""
Secondary index on actor
"""
creator = UnicodeAttribute(hash_key=True)
[docs]
class Actor(Model):
"""
DynamoDB data model for an actor
"""
id = UnicodeAttribute(hash_key=True)
creator = UnicodeAttribute()
passphrase = UnicodeAttribute()
creator_index = CreatorIndex()
[docs]
class DbActor:
"""
DbActor does all the db operations for actor objects
"""
[docs]
def get(self, actor_id: str | None = None) -> dict[str, Any] | None:
"""Retrieves the actor from the database"""
if not actor_id:
return None
try:
self.handle = Actor.get(actor_id, consistent_read=True)
except DoesNotExist:
return None
except Exception as e:
# Anything that is not a confirmed absence — a throttle, a timeout,
# bad credentials, a missing table — used to return None here with
# no log line at any level, so callers could not tell "no such
# actor" from "the read failed" and nothing recorded that it had.
# None is still returned for backward compatibility (raising would
# turn every throttle across auth, OAuth2 and MCP into a 500), so
# this ERROR is the only signal that an existence check just
# answered "no" for an infrastructure reason. Callers needing the
# distinction should use actingweb.deletion.get_deletion_status(),
# which reports UNKNOWN rather than guessing.
logger.error(
f"Failed to read actor {actor_id} from DynamoDB: "
f"{type(e).__name__}: {e}. Returning None — callers will see "
"this as 'actor does not exist'."
)
return None
if self.handle:
t = self.handle
return {
"id": t.id,
"creator": t.creator,
"passphrase": t.passphrase,
}
else:
return None
[docs]
def get_by_creator(
self, creator: str | None = None
) -> dict[str, Any] | list[dict[str, Any]] | None:
"""Retrieves the actor from db based on creator field
Returns None if none was found. If one is found, that one is
loaded in the object. If more, all are returned.
"""
if not creator:
return None
if "@" in creator:
creator = creator.lower()
self.handle = Actor.creator_index.query(creator)
ret = []
for c in self.handle:
logger.debug("get_by_creator matched actor id=" + c.id)
ret.append(self.get(actor_id=c.id))
return ret
[docs]
def modify(
self, creator: str | None = None, passphrase: bytes | None = None
) -> bool:
"""Modify an actor"""
if not self.handle:
logger.debug("Attempted modification of DbActor without db handle")
return False
if creator and len(creator) > 0:
# Email in creator needs to be lower case
if "@" in creator:
creator = creator.lower()
self.handle.creator = creator # type: ignore[attr-defined]
if passphrase and len(passphrase) > 0:
self.handle.passphrase = passphrase.decode("utf-8") # type: ignore[attr-defined]
self.handle.save() # type: ignore[attr-defined]
return True
[docs]
def create(
self,
actor_id: str | None = None,
creator: str | None = None,
passphrase: str | None = None,
) -> bool:
"""Create a new actor"""
if not actor_id:
return False
if not creator:
creator = ""
# Email in creator needs to be lower case
if "@" in creator:
creator = creator.lower()
if not passphrase:
passphrase = ""
if self.get(actor_id=actor_id):
logger.warning("Trying to create actor that exists(" + actor_id + ")")
return False
self.handle = Actor(id=actor_id, creator=creator, passphrase=passphrase)
self.handle.save()
return True
[docs]
def delete(self):
"""Deletes the actor in the database"""
if not self.handle:
logger.debug("Attempted delete of DbActor without db handle")
return False
self.handle.delete() # type: ignore[attr-defined]
self.handle = None
return True
def __init__(self):
self.handle = None
ensure_table(Actor)
[docs]
class DbActorList:
"""
DbActorList does all the db operations for list of actor objects
"""
[docs]
def fetch(self):
"""Retrieves ALL actors in the database.
Admin/maintenance use only: this is a deliberate full-table Scan,
O(table size) and unpaginated — do not call it on a serving path.
"""
self.handle = Actor.scan()
if self.handle:
ret = []
for t in self.handle:
ret.append(
{
"id": t.id,
"creator": t.creator,
}
)
return ret
else:
return False
def __init__(self):
self.handle = None