Skip to content

Commit

Permalink
feat: support easily injectable dynamodb resource to use for caching.
Browse files Browse the repository at this point in the history
  • Loading branch information
joshorr committed Jan 2, 2024
1 parent 7879a1a commit 534bf00
Showing 1 changed file with 29 additions and 11 deletions.
40 changes: 29 additions & 11 deletions xcon/providers/dynamo.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,13 +6,15 @@
import os
import random
from collections import defaultdict
from typing import Dict, Optional, Callable, Iterable, Sequence, Tuple
from typing import Dict, Optional, Callable, Iterable, Sequence, Tuple, Any
from typing import Mapping

from boto3.dynamodb import conditions
from xboto.resource import dynamodb
from xboto.dependencies import BotoResources
from xinject import Dependency
from xsentinels import Default
from xloop import xloop
from xsentinels.default import DefaultType

from xcon.directory import Directory, DirectoryListing, DirectoryOrPath, DirectoryItem, \
DirectoryChain
Expand Down Expand Up @@ -294,7 +296,7 @@ def get_item(
"""
# Cache needs all of this stuff to do proper caching.
environ = self._get_environ_to_use(environ)
if not directory or not directory_chain or not environ:
if not directory_chain or not environ:
return None

return self._get_listing(
Expand Down Expand Up @@ -395,7 +397,29 @@ def _get_environ_to_use(
return Directory(service=e_service, env=e_env)


# Most of this code could be shared from `xyn_model_dynamo`, but we don't want to import that
class DynamoDBResource(Dependency):
dynamodb_xboto_resource: BotoResources.DynamoDB | DefaultType = Default
_table_name_to_boto_resource: dict[str, Any]

def __init__(self, dynamodb_xboto_resource: BotoResources.DynamoDB | DefaultType = Default):
self.dynamodb_xboto_resource = dynamodb_xboto_resource
self._table_name_to_boto_resource = {}

def table_resource(self, table_name):
if resource := self._table_name_to_boto_resource.get(table_name):
return resource

if self.dynamodb_xboto_resource is Default:
dynamodb = BotoResources.grab().dynamodb
else:
dynamodb = self.dynamodb_xboto_resource.boto_resource

resource = dynamodb.Table(table_name)
self._table_name_to_boto_resource[table_name] = resource
return resource


# Most of this code could be shared from `xmodel_dynamo`, but we don't want to import that
# in this library (it's a bit heavy). So for now, we are duplicating some of that functionality
# for use here, in a much simpler (but WAY less feature-rich) way:
class _ConfigDynamoTable:
Expand All @@ -422,13 +446,7 @@ def table(self):
""" DynamoDB table resource.
We lazily get the resource, so we don't have to verify/create it if not needed.
"""
table = self._table
if table is not None:
return table

table = dynamodb.Table(self.table_name)
self._table = table
return table
return DynamoDBResource.grab().table_resource(self.table_name)

def __init__(
self,
Expand Down

0 comments on commit 534bf00

Please sign in to comment.