I'm using Cloud Composer, and I have a DAG that has one task that calls an HTTPS-triggered cloud function that sends out an email (due to restrictions on the project I'm working on, I had to do it this way). The most simple form of this works. I can trigger the cloud function, and the emails are being sent successfully. However, I want to pass some variables I'm defining in the DAG to the Cloud Function, and this is where something is failing. I was using the usual way to pass parameters to the request URL.
This was the way I was defining the DAG:
# --------------------------------------------------------------------------------
# Import Libraries
# --------------------------------------------------------------------------------
import datetime
from airflow.models import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from airflow.contrib.operators.bigquery_operator import BigQueryOperator
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator,BigQueryExecuteQueryOperator
from airflow.providers.google.common.utils import id_token_credentials as id_token_credential_utils
import google.auth.transport.requests
from google.auth.transport.requests import AuthorizedSession
# --------------------------------------------------------------------------------
# Set variables
# --------------------------------------------------------------------------------
(...)
report_name_url = "report_name_url"
end_user = "end_user@email.com"
# --------------------------------------------------------------------------------
# Functions
# --------------------------------------------------------------------------------
def invoke_cloud_function():
url = "https://<trigger_url>?report_name_url={}&end_user={}".format(report_name_url, end_user) #I'M ADDING THE STRINGS TO THE URL AFTER THE ?, TO PASS WHAT I WANT AS ARGUMENTS TO THE CLOUD FUNCTION
request = google.auth.transport.requests.Request() #this is a request for obtaining the the credentials
id_token_credentials = id_token_credential_utils.get_default_id_token_credentials(url, request=request) # If your cloud function url has query parameters, remove them before passing to the audience
resp = AuthorizedSession(id_token_credentials).request("GET", url=url) # the authorized session object is used to access the Cloud Function
print(resp.status_code) # should return 200
print(resp.content) # the body of the HTTP response
# --------------------------------------------------------------------------------
# Define DAG
# --------------------------------------------------------------------------------
with DAG(
dag_id,
schedule_interval= '0 13 05 * *', # DAG Cron scheduler
default_args = default_args) as dag:
(...)
send_email = PythonOperator(
task_id="send_email",
python_callable=invoke_cloud_function
)
start >> run_stored_procedure >> composer_logging >> send_email >> end
This is what I have as far as the DAG goes. From the perspective of the cloud function, I have the following:
def send_email(request):
import ssl
from email.message import EmailMessage
import smtplib
import os
report_name_url = request.args.get('report_name_url')
report_name = report_name_url.replace("_", " ")
end_user = request.args.get('end_user')
(...)
context = ssl.create_default_context()
with smtplib.SMTP_SSL('smtp.gmail.com', 465, context=context) as smtp:
smtp.login(sender_email, password)
smtp.sendmail(sender_email, receiver_email, em.as_string())
Can someone point me toward a solution for my use-case?
Thank you very much.
Edit for added context:
I'm getting the following information from the logs:
"(...)Unauthorized</h1>\n<h2>Your client does not have permission to the requested URL <code>...</code>.</h2>\n<h2></h2>\n</body></html>\n'"
This is odd because I think this was the error I was getting BEFORE giving permissions to my service account to invoke the cloud function. Now, that permission is in place.
The only problem I foresee is that I'm not exactly calling the original URL that triggers the cloud function, since I'm adding parameters. Could this be the problem?