Skip to content

Instantly share code, notes, and snippets.

@em-shea
Created January 16, 2022 19:25
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save em-shea/da2ef1080df810910e18ccdf6f4960e1 to your computer and use it in GitHub Desktop.
Save em-shea/da2ef1080df810910e18ccdf6f4960e1 to your computer and use it in GitHub Desktop.
An example of a user service querying DynamoDB and formatting the response
import os
import boto3
import datetime
from typing import List, Dict
from dataclasses import dataclass
from boto3.dynamodb.conditions import Key
@dataclass
class Subscription:
list_name: str
unique_list_id: str
list_id: str
character_set: str
status: str
date_subscribed: str
@dataclass
class User:
email_address: str
user_id: str
character_set_preference: str
date_created: datetime.datetime
user_alias: str
user_alias_pinyin: str
user_alias_emoji: str
subscriptions: List[Subscription]
table = boto3.resource('dynamodb', region_name=os.environ['AWS_REGION']).Table(os.environ['TABLE_NAME'])
def get_single_user(cognito_id):
user_data = query_single_user(cognito_id)
response = _format_user_data(user_data)
return response
def query_single_user(cognito_id):
user_key = "USER#" + cognito_id
response = table.query(KeyConditionExpression=Key('PK').eq(user_key))
return response['Items']
def _format_user_data(user_data):
user = None
subscription_list = []
# Loop through all users and subs
for item in user_data:
# If Dynamo item is user metadata, create User class
if 'Email address' in item:
user = User(
email_address = item['Email address'],
user_id = item['PK'][5:],
character_set_preference = item['Character set preference'],
date_created = item['Date created'],
user_alias = item['User alias'],
user_alias_pinyin = item['User alias pinyin'],
user_alias_emoji = item['User alias emoji'],
subscriptions = []
)
# If Dynamo item is a list subscription, add the list to the user's lists dict
if 'List name' in item:
# Shortening list id from unique id (ex, LIST#12345#SIMPLIFIED) since we're already returning character_set
if 'SIMPLIFIED' in item['SK']:
list_id = item['SK'][5:-11]
if 'TRADITIONAL' in item['SK']:
list_id = item['SK'][5:-12]
sub = Subscription(
list_name = item['List name'],
unique_list_id = item['SK'][5:],
list_id = list_id,
character_set = item['Character set'],
status = item['Status'],
date_subscribed = item['Date subscribed']
)
subscription_list.append(sub)
# Sort lists by list id to appear in order (Level 1, Level 2, etc.)
subscription_list = sorted(subscription_list, key=lambda k: k.list_id, reverse=False)
user.subscriptions = subscription_list
return user
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment