Service Ridership Dashboard Queries

This page documents query functions in the Service Ridership Dashboard module.

ingestor.chalicelib.service_ridership_dashboard.queries

query_scheduled_service(start_date, end_date, route_id)

Query the ScheduledServiceDaily DynamoDB table for a route within a date range.

Parameters:

Name	Type	Description	Default
start_date	date	The start date of the query range.	required
end_date	date	The end date of the query range.	required
route_id	str	The MBTA route identifier to query.	required

Returns:

Type	Description
list[ScheduledServiceRow]	A list of ScheduledServiceRow dicts from DynamoDB.

Source code in ingestor/chalicelib/service_ridership_dashboard/queries.py

def query_scheduled_service(start_date: date, end_date: date, route_id: str) -> list[ScheduledServiceRow]:
    """Query the ScheduledServiceDaily DynamoDB table for a route within a date range.

    Args:
        start_date: The start date of the query range.
        end_date: The end date of the query range.
        route_id: The MBTA route identifier to query.

    Returns:
        A list of ScheduledServiceRow dicts from DynamoDB.
    """
    table = dynamodb.Table("ScheduledServiceDaily")
    date_condition = Key("date").between(start_date.isoformat(), end_date.isoformat())
    route_condition = Key("routeId").eq(route_id)
    condition = date_condition & route_condition
    response = table.query(KeyConditionExpression=condition)
    return ddb_json.loads(response["Items"])

query_ridership(start_date, end_date, line_id)

Query the Ridership DynamoDB table for a line within a date range.

Parameters:

Name	Type	Description	Default
start_date	date	The start date of the query range.	required
end_date	date	The end date of the query range.	required
line_id	str	The MBTA line identifier to query.	required

Returns:

Type	Description
list[RidershipRow]	A list of RidershipRow dicts from DynamoDB.

Source code in ingestor/chalicelib/service_ridership_dashboard/queries.py

def query_ridership(start_date: date, end_date: date, line_id: str) -> list[RidershipRow]:
    """Query the Ridership DynamoDB table for a line within a date range.

    Args:
        start_date: The start date of the query range.
        end_date: The end date of the query range.
        line_id: The MBTA line identifier to query.

    Returns:
        A list of RidershipRow dicts from DynamoDB.
    """
    table = dynamodb.Table("Ridership")
    date_condition = Key("date").between(start_date.isoformat(), end_date.isoformat())
    line_condition = Key("lineId").eq(line_id)
    condition = date_condition & line_condition
    response = table.query(KeyConditionExpression=condition)
    return ddb_json.loads(response["Items"])