in backend_apis/app/main.py [0:0]
def post_audiences(data: AudiencesRequest) -> AudiencesResponse:
"""Transform a question in NL to SQL and query BQ.
Parameters:
question: Question to be asked to BQ.
Returns:
audiences: dict with emails
gen_code: SQL code
"""
# Audiences
tag_template_name = (f'projects/{project_id}/locations/'
f'{location}/tagTemplates/{config["global"]["tag_name"]}')
query_metadata = (
f'SELECT * FROM `{project_id}.{dataset_id}.INFORMATION_SCHEMA.TABLES`'
' WHERE table_name NOT LIKE "%metadata%"')
try:
audiences, gen_code, prompt = utils_codey.generate_sql_and_query(
llm=code_llm,
datacatalog_client=datacatalog_client,
prompt_template=prompt_nl_sql,
query_metadata=query_metadata,
question=data.question,
project_id=project_id,
dataset_id=dataset_id,
tag_template_name=tag_template_name,
bqclient=bq_client
)
crm_data = bulk_email_util.generate_information(audiences).to_dict('records')
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
return AudiencesResponse(
audiences={"data":audiences,"crm_data":crm_data},
gen_code=gen_code,
prompt=prompt
)