1import time
2from datetime import datetime, timedelta, timezone
3
4import boto3
5
6
7client = boto3.client("logs", region_name="us-east-1")
8
9end_time = int(datetime.now(timezone.utc).timestamp())
10start_time = int((datetime.now(timezone.utc) - timedelta(hours=1)).timestamp())
11
12query = """
13fields @timestamp, @message, @logStream
14| filter @message like /ERROR/ | sort @timestamp desc | limit 20 """ response = client.start_query( logGroupName="/aws/lambda/my-function", startTime=start_time, endTime=end_time, queryString=query, ) query_id = response["queryId"] print("started query:", query_id) ``` The query language is CloudWatch Logs Insights syntax. You can filter, parse fields, sort, aggregate, and limit results without downloading the whole log stream first. Before running this in production code, make sure the caller has `logs:StartQuery` and `logs:GetQueryResults` permissions for the target log group. It is also worth setting the AWS region explicitly on the client, because many empty-result bugs turn out to be queries sent to the wrong region rather than problems with the query itself. ## Poll Until Results Are Ready `start_query` is asynchronous, so the next step is polling `get_query_results` until the status becomes `Complete`, `Failed`, `Cancelled`, or `Timeout`. ```python import time def wait_for_query(client, query_id: str, delay: float = 1.0): while True: result = client.get_query_results(queryId=query_id) status = result["status"] if status == "Complete": return result["results"] if status in ("Failed", "Cancelled", "Timeout"): raise RuntimeError(f"query ended with status: {status}") time.sleep(delay) rows = wait_for_query(client, query_id) print("row count:", len(rows)) ``` Each row is returned as a list of field records. That shape is awkward at first, so most code converts each row into a normal Python dictionary. ```python def normalize_rows(rows): normalized = [] for row in rows: item = {} for field in row: item[field["field"]] = field["value"] normalized.append(item) return normalized for item in normalize_rows(rows): print(item["@timestamp"], item.get("@message", "")) ``` ## Use `filter_log_events` for Direct Event Reads If you do not need Logs Insights syntax and just want raw events that match a simple filter pattern, `filter_log_events` is sometimes simpler. ```python import boto3 client = boto3.client("logs", region_name="us-east-1") response = client.filter_log_events( logGroupName="/aws/lambda/my-function", filterPattern="ERROR", limit=10, ) for event in response["events"]: print(event["timestamp"], event["message"]) ``` This API is good for quick retrieval, but it is less expressive than Logs Insights for grouped analysis or field extraction. If you plan to page through many raw events, check the response for `nextToken` and keep requesting pages until it disappears. For interactive scripts, Logs Insights is usually a better fit because it lets AWS do more of the filtering and sorting work before the data reaches your code. ## Common Pitfalls - Passing milliseconds instead of seconds to `start_query`. That usually returns an empty or confusing time window. - Forgetting to poll `get_query_results`. The initial call only creates the query job. - Ignoring AWS region settings. Log groups are regional, so the client must target the correct region. - Expecting `get_query_results` to return plain dictionaries. It returns lists of field-value records that you usually need to normalize. - Polling too aggressively in a tight loop. Add a short delay to reduce unnecessary API calls. ## Summary - Use `start_query` and `get_query_results` for CloudWatch Logs Insights queries from Python. - Pass the time range as Unix seconds and specify the correct log group and region. - Poll until the query status becomes `Complete`. - Normalize the result rows into dictionaries before further processing. - Use `filter_log_events` when you want straightforward event retrieval without Logs Insights syntax.