Created
January 16, 2022 19:25
-
-
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
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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