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.