AWS DynamoDB table.query() not working as expected
I'm working with a fast-api backend where I've set it up to work with AWS's DynamoDB client. I've declared the following wrapper that helps me manage read and writes from the database.
(A self defined module called database holds two files, the first defines a class called DeviceDataManagaer and the second has an init).
class DeviceDataManager():
def __init__(self, dyn_resource):
"""
: param dyn_resource: A Boto3 DynamoDB resource.
"""
self.dyn_resource = dyn_resource
self.device_table = None
self.master_order_table = None
self.master_history_table = None
def load_tables(
self,
device_table: str,
master_order_table:str,
master_history_table:str
) -> bool:
"""
Attempts to load the given tables, storing them in a disctionary that is stored
as a member variable. Returns a boolean indicating whether all tables were
loaded or not.
"""
table_names = (device_table, master_order_table, master_history_table)
table_existence = [False] * len(table_names)
loading_tables = []
for i, table_in in enumerate(table_names):
try:
table = self.dyn_resource.Table(table_in)
table.load()
table_existence[i] = True
except ClientError as err:
if err.response["Error"]["Code"] == "ResourceNotFoundException":
table_existence[i] = False
else:
logger.error(
"Couldn't check for existence of tables. Here's why: %s: %s",
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
loading_tables.append(table)
self.device_table = loading_tables[0]
self.master_order_table, self.master_history_table = loading_tables[1:3]
return all(table_existence)
def get_device_data(self, serial: str) -> dict | None:
"""
Gets an item from the Device Data Table
"""
try:
# response = self.device_table.get_item(Key={"Serial_Number": serial, "Local_Time_Str": "string"})
# FIXME: Replace arg to Serial_Number
if True:
print(repr(serial))
response = self.device_table.query(
KeyConditionExpression=(
Key('Serial_Number').eq(serial)
)
)
except ClientError as err:
logger.error(
"Couldn't get item from %s. Here's why: %s: %s",
self.device_table.name,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return response.get("Item")
# ... More methods here...
I have three tables in the database, where the _init_ looks like the following:
import os, pathlib, boto3
from dotenv import load_dotenv
from .DeviceDataManager import DeviceDataManager
base_dir = pathlib.Path("June Presentation/app").parent.parent.parent
# Load DynamoDB Credentials
key_path = "./app/environment vars/aws.env"
if os.path.exists(base_dir.joinpath(key_path)):
load_dotenv(base_dir.joinpath(key_path))
else:
raise RuntimeError(f"Credentials not found for DeviceDataManager's DynamoDB Client at {base_dir.joinpath(key_path)}")
class __Config:
DB_REGION_NAME = os.getenv('DB_REGION_NAME')
DB_ACCESS_KEY_ID = os.getenv('DB_ACCESS_KEY_ID')
DB_SECRET_ACCESS_KEY = os.getenv('DB_SECRET_ACCESS_KEY')
def get_device_db(
device_table: str = "Device_Data",
order_table: str = "Master_Order",
history_table: str = "Master_History"
) -> DeviceDataManager:
"""
Creates an instance of DeviceDataManager allowing access to the DynamoDB's table.
"""
db = DeviceDataManager(
boto3.resource(
'dynamodb',
region_name=__Config.DB_REGION_NAME,
aws_access_key_id=__Config.DB_ACCESS_KEY_ID,
aws_secret_access_key=__Config.DB_SECRET_ACCESS_KEY
)
)
if (not db.load_tables(device_table, order_table, history_table)):
raise FileNotFoundError("One or more tables not found!")
return db
The Device_Data table has the following model:
from pydantic import BaseModel
from pydantic import BaseModel, conlist
from decimal import Decimal
class Device(BaseModel):
Serial_Number: str
class DeviceData(Device):
Local_Time_Str: str # Sort Key
local_ip: str
touch_mode: int
location: str
region: str
country: str
latitude: str
longitude: str
temperature: Decimal
condition: str
wind_speed: Decimal
humidity: Decimal
fan_status_arr: conlist(int, min_length=5, max_length=5)
fan_selecting_pwm: int
The other two Master Tables work fine, but the Device_Data table has me plucking my hair out. It has a Parrtition_Key called "Serial_Number" and a sort key called "Local_Time_Str". I have tested the connection with the db, everything else works fine. When I call device_data.get_item with a partition and sort key, it also works but for some reason when I use table.query() with only the partition key I get a 404 Not Found response, even though when I try it on the AWS Console (on the website it works).
AWS Console:
Response on Backend:
The HTTP GET method is as follows (for fast-api router)
from fastapi import APIRouter, Depends, HTTPException
from ..models.models import DeviceData, MasterData
from ..database import get_device_db, DeviceDataManager
from ..internal.Authentication import get_current_active_user
from ..models.Authentication import User
from typing import Annotated
router = APIRouter(
prefix="/device",
tags=["device"],
responses={404: {"description": "Not found"}},
)
# Instantiate the DynamoDB Table Manager (For Device Data)
try:
db: DeviceDataManager = get_device_db()
except FileNotFoundError as err:
raise print(err)
@router.get("/get-device", response_model=DeviceData)
async def get_item(
serial: str,
current_user: Annotated[User, Depends(get_current_active_user)]
) -> DeviceData | None:
print("NotImplementedWarning!") #FIXME
item = db.get_device_data(serial)
print(item) # DEBUG
if item:
return item
raise HTTPException(status_code=404, detail="Item not found")
I would appreciate any help with resolving this issue.
I tried the table.get_item() with the partition key and sort key, which worked. I also tried the query on the AWS DDB console which also worked. I read the documentation for AWS DDB and other articles online but nothing stood out or helped me in any way.
As pointed out by @jarmod in the comments, the issue is quite simple and a basic typo. Replacing "Item" with "Items" fixes the issue.
def get_device_data(self, serial: str) -> dict | None:
"""
Gets an item from the Device Data Table
"""
try:
# response = self.device_table.get_item(Key={"Serial_Number": serial, "Local_Time_Str": "string"})
# FIXME: Replace arg to Serial_Number
if True:
print(repr(serial))
response = self.device_table.query(
KeyConditionExpression=(
Key('Serial_Number').eq(serial)
)
)
except ClientError as err:
logger.error(
"Couldn't get item from %s. Here's why: %s: %s",
self.device_table.name,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return response.get("Item")