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.

New contributor

Shaheer ziya is a new contributor to this site. Take care in asking for clarification, commenting, and answering.
Check out our Code of Conduct.

Trang chủ Giới thiệu Sinh nhật bé trai Sinh nhật bé gái Tổ chức sự kiện Biểu diễn giải trí Dịch vụ khác Trang trí tiệc cưới Tổ chức khai trương Tư vấn dịch vụ Thư viện ảnh Tin tức - sự kiện Liên hệ Chú hề sinh nhật Trang trí YEAR END PARTY công ty Trang trí tất niên cuối năm Trang trí tất niên xu hướng mới nhất Trang trí sinh nhật bé trai Hải Đăng Trang trí sinh nhật bé Khánh Vân Trang trí sinh nhật Bích Ngân Trang trí sinh nhật bé Thanh Trang Thuê ông già Noel phát quà Biểu diễn xiếc khỉ Xiếc quay đĩa Dịch vụ tổ chức sự kiện 5 sao Thông tin về chúng tôi Dịch vụ sinh nhật bé trai Dịch vụ sinh nhật bé gái Sự kiện trọn gói Các tiết mục giải trí Dịch vụ bổ trợ Tiệc cưới sang trọng Dịch vụ khai trương Tư vấn tổ chức sự kiện Hình ảnh sự kiện Cập nhật tin tức Liên hệ ngay Thuê chú hề chuyên nghiệp Tiệc tất niên cho công ty Trang trí tiệc cuối năm Tiệc tất niên độc đáo Sinh nhật bé Hải Đăng Sinh nhật đáng yêu bé Khánh Vân Sinh nhật sang trọng Bích Ngân Tiệc sinh nhật bé Thanh Trang Dịch vụ ông già Noel Xiếc thú vui nhộn Biểu diễn xiếc quay đĩa Dịch vụ tổ chức tiệc uy tín Khám phá dịch vụ của chúng tôi Tiệc sinh nhật cho bé trai Trang trí tiệc cho bé gái Gói sự kiện chuyên nghiệp Chương trình giải trí hấp dẫn Dịch vụ hỗ trợ sự kiện Trang trí tiệc cưới đẹp Khởi đầu thành công với khai trương Chuyên gia tư vấn sự kiện Xem ảnh các sự kiện đẹp Tin mới về sự kiện Kết nối với đội ngũ chuyên gia Chú hề vui nhộn cho tiệc sinh nhật Ý tưởng tiệc cuối năm Tất niên độc đáo Trang trí tiệc hiện đại Tổ chức sinh nhật cho Hải Đăng Sinh nhật độc quyền Khánh Vân Phong cách tiệc Bích Ngân Trang trí tiệc bé Thanh Trang Thuê dịch vụ ông già Noel chuyên nghiệp Xem xiếc khỉ đặc sắc Xiếc quay đĩa thú vị
Trang chủ Giới thiệu Sinh nhật bé trai Sinh nhật bé gái Tổ chức sự kiện Biểu diễn giải trí Dịch vụ khác Trang trí tiệc cưới Tổ chức khai trương Tư vấn dịch vụ Thư viện ảnh Tin tức - sự kiện Liên hệ Chú hề sinh nhật Trang trí YEAR END PARTY công ty Trang trí tất niên cuối năm Trang trí tất niên xu hướng mới nhất Trang trí sinh nhật bé trai Hải Đăng Trang trí sinh nhật bé Khánh Vân Trang trí sinh nhật Bích Ngân Trang trí sinh nhật bé Thanh Trang Thuê ông già Noel phát quà Biểu diễn xiếc khỉ Xiếc quay đĩa
Thiết kế website Thiết kế website Thiết kế website Cách kháng tài khoản quảng cáo Mua bán Fanpage Facebook Dịch vụ SEO Tổ chức sinh nhật