Cloud Trail Query Test

cloudtrail_4.py

query_text = f"""
SELECT eventname, substr(eventtime, 1, 10) as eventdate,count(*)
FROM {database}.{cloudtrail_table}
WHERE region='{region}' 
and year=date_format(date_add('day',-1,current_date),'%Y') 
and month=date_format(date_add('day',-1,current_date),'%m')
and day=date_format(date_add('day',-1,current_date),'%d')
and eventname='StartQueryExecution'
GROUP BY 1, 2
ORDER BY 1, 2
"""

athena = my_session.client('athena', region_name=region)

response = athena.create_named_query(
    Name='Athena QLI Test Query',
    Description='Developer: [email protected]',
    Database=database,
    QueryString=query_text,
    ClientRequestToken=f"{hash(query_text):32}"
)

import uuid

my_query_id = response['NamedQueryId']

print(f"Saved test query with ID {my_query_id}")

response = athena.start_query_execution(
    QueryString=query_text,
    ClientRequestToken=str(uuid.uuid1()),
    QueryExecutionContext={
        'Database': database
    },
    ResultConfiguration={
        'OutputLocation': results,
    }
)

my_query_execution_id = response['QueryExecutionId']

for j in range(10):
    try:
        print(f"Get execution results for {my_query_execution_id}")
        response = athena.get_query_results(
            QueryExecutionId=my_query_execution_id
        )
        break
    except:
        time.sleep(10)

my_result = response.get('ResultSet').get('Rows')
print(f"Your query ran and produced: {my_result}")