Module xdynamo.db
Used to keep track of a shared connection and Dynamo table resources.
If you want to easily send/get information from Dynamo via orm.
Important Classes
-
DynamoDB
: A resource (seexyn_resource
for more details) that keeps track of a shared boto3 dynamo session/connection. It also caches the fact if a table is created (ie: no need to check every time, just the first time we use a table). -
DynamoTableCreator
: Interface used byDynamoDB
to have an easy way to initiate the creation of a Dynamo table if needed (if it does not exist).
Classes
class DynamoDB
-
Resource that represents a DynamoDB connection. This allows us to pool the dynamo connection among everything that uses Dynamo, so that we can reuse existing connections. This speeds up Dynamo by a fair amount. Every time you open a connection with Dynamo, it has to figure out how to get the encryption key, but subsequent requests using the same connection don't have to.
Expand source code
class DynamoDB(Dependency, attributes_to_skip_while_copying=["_table", "_verified", "_db"]): """ Resource that represents a DynamoDB connection. This allows us to pool the dynamo connection among everything that uses Dynamo, so that we can reuse existing connections. This speeds up Dynamo by a fair amount. Every time you open a connection with Dynamo, it has to figure out how to get the encryption key, but subsequent requests using the same connection don't have to. """ _tables: Dict[str, Any] _verified: Dict[str, bool] def __init__(self): self._tables = {} self._verified = {} @property def db(self): return dynamodb def table(self, name: str, table_creator: Optional[DynamoTableCreator] = None): """Returns existing table or creates + returns one if we don't have the resource currently. When we create a new one, we will remember it in a weak-fashion in case we get asked again. Once all other references to the table resource are gone the python garbage-collector will cleanup our reference automatically. If that happens and we get asked for that table_resource, we will create and return a new one [and remember if for future use]. The reason to provide a table_creator is so you can easily Mock DynamoDB, as it will be created in the 'Mock' framework automatically. Also when devs run the code locally without deploying it, it will create the table for them. When you deploy code via serverless, it should be the thing the creates the table for the deployed environment [ie: testing/prod/etc]. The table should already exist that way when the app runs, and so it does **NOT** need table-creating AWS permissions. Args: name: Name of the table to get a resource for. table_creator (Union[DynamoTableCreator, None]): ... If None: Won't verify table is created or ready. If DynamoTableCreator: Verifies the table exists and is ready. If it does not exist we create the table by 'calling' whats passed in here and wait for it to be ready. If the table is in a status that indicates it can't be used at the moment [example: If table is 'DELETING'], we raise an XynLibError. """ verified = self._verified.get(name, False) try: from xcon import xcon_settings # We only verify/create-table-if-needed in specific environments. if xcon_settings.environment not in _auto_create_table_only_in_environments: verified = True except ImportError: # If `xcon` unavailable, just assume we don't want to auto-create tables # todo: Put in a configurable setting that allows one to turn on/off # auto-table-creation. # (and some way to communicate billing mode???). # # todo: Log about why not creating tables, but log it only once. verified = True table = self._tables.get(name) if table is not None and verified: return table if table is None: table = dynamodb.Table(name) if verified: self._tables[name] = table return table if not table_creator: # Don't verify table if we don't have a table creator, just return it. self._tables[name] = table self._verified[name] = False return table try: log.info(f"Getting Table Status for ({name}).") status = table.table_status # It turns out, a table is still usable while "UPDATING", so only worry about # DELETING and CREATING. if status == "CREATING": log.warning( f"Dynamo status for table ({name}) is CREATING; based on past experience " f"we can still use the table [to at least read values] if the table is " f"being restored from backup while it's in the CREATING status. So I am not " f"going to wait for table to become ACTIVE, I'll try to use the table " f"immediately." ) # This is how we could wait for the table, disabling for now, see log.warning ^^^ # table.wait_until_exists() elif status == "DELETING": raise XModelDynamoError(f"Dynamo Table ({name}) status is 'DELETING'???") except ClientError: # This means the table has not been created yet, we create and wait for it to exist. log.warning(f"Dynamo table ({name}) does not exist, creating...") table = table_creator(dynamo=self) table.wait_until_exists() self._tables[name] = table self._verified[name] = True return table
Ancestors
- xinject.dependency.Dependency
Instance variables
prop db
-
Expand source code
@property def db(self): return dynamodb
Methods
def table(self, name: str, table_creator: Optional[DynamoTableCreator] = None)
-
Returns existing table or creates + returns one if we don't have the resource currently. When we create a new one, we will remember it in a weak-fashion in case we get asked again. Once all other references to the table resource are gone the python garbage-collector will cleanup our reference automatically. If that happens and we get asked for that table_resource, we will create and return a new one [and remember if for future use].
The reason to provide a table_creator is so you can easily Mock DynamoDB, as it will be created in the 'Mock' framework automatically.
Also when devs run the code locally without deploying it, it will create the table for them.
When you deploy code via serverless, it should be the thing the creates the table for the deployed environment [ie: testing/prod/etc]. The table should already exist that way when the app runs, and so it does NOT need table-creating AWS permissions.
Args
name
- Name of the table to get a resource for.
table_creator
:Union[DynamoTableCreator, None]
-
… If None: Won't verify table is created or ready.
If DynamoTableCreator: Verifies the table exists and is ready. If it does not exist we create the table by 'calling' whats passed in here and wait for it to be ready. If the table is in a status that indicates it can't be used at the moment [example: If table is 'DELETING'], we raise an XynLibError.
class DynamoTableCreator
-
A callable that can, when called, create the needed Dynamo table, and then return the newly created table as a boto3 table resource [via: return dynamo.db.create_table(…)].
Expand source code
class DynamoTableCreator: """A callable that can, when called, create the needed Dynamo table, and then return the newly created table as a boto3 table resource [via: return dynamo.db.create_table(...)]. """ def __call__(self, dynamo: DynamoDB) -> Any: """Something like this:: return dynamo.db.create_table(TableName="SomeName", KeySchema=[...], ...) """ raise NotImplementedError("Must Implement To Create Dynamo Table.")