[Python] How to pass CSV file between Airflow tasks

I have this code (it's working) and I want to break it down to 2 functions (Airflow tasks).

1. Pull Json file and parse it to CSV
2. Upload CSV to S3 bucket.

I can't pass the CSV file between the Airflow tasks (dags) I tried xcom push and pull but I'm missing something.

I know it's not the best way but the file is very small and I will learn it in time.

import os
import csv
import firebase_admin
from firebase_admin import credentials, auth
import boto3
from datetime import datetime

import json
from airflow.models import Variable


def fetch_all_users():
#AWS Cred
aws_access_key_id = Variable.get('RIVERY_AWS_ACCESS_KEY_ID')
aws_secret_access_key = Variable.get('RIVERY_AWS_SECRET_ACCESS_KEY')
bucket = Variable.get('RIVERY_BUCKET')
firebase_key = 'firebase/key/firebase.json'

#Starting S3 session
session = boto3.Session(aws_access_key_id=aws_access_key_id, aws_secret_access_key=aws_secret_access_key)
s3 = session.resource('s3')

#Getting Firebase key and parse it
obj = s3.Object(bucket, firebase_key)
file_content = obj.get()['Body'].read().decode('utf-8')
json_data = json.loads(file_content)

#init Firebase connection
cred = credentials.Certificate(json_data)
firebase_admin.initialize_app(cred)

# List all users
all_users = auth.list_users()

# Extract user data with provider data
users_data = []
for user in all_users.iterate_all():
user_data = {
'uid': user.uid,
'email': user.email,
'display_name': user.display_name,
'creation_timestamp': user.user_metadata.creation_timestamp, # Convert to ISO format
'provider_data': [{
'provider_id': provider.provider_id,
'email': provider.email,
'phone_number': provider.phone_number,
'photo_url': provider.photo_url,
'uid': provider.uid,
'last_sign_in_timestamp': user.user_metadata.last_sign_in_timestamp
} for provider in user.provider_data],
# Add other user properties as needed
}
users_data.append(user_data)

#Parse json to CSV
today = datetime.now().strftime("%Y-%m-%d")
csv_filename = f"firebase_users_s3_(today).csv"
# CSV header
csv_fields = ['uid', 'email', 'display_name', 'creation_timestamp', 'provider_id', 'provider_email', 'provider_phone_number',
'provider_photo_url', 'provider_uid', 'last_sign_in_timestamp']

# Writing to csv file
with open(csv_filename, mode='w', newline='') as csv_file:
writer = csv.DictWriter(csv_file, fieldnames=csv_fields)

# Write header
writer.writeheader()

# Write user data
for user in users_data:
for provider_data in user['provider_data']:
user_row = {
'uid': user['uid'],
'email': user['email'],
'display_name': user['display_name'],
'creation_timestamp': user['creation_timestamp'],
'provider_id': provider_data['provider_id'],
'provider_email': provider_data['email'],
'provider_phone_number': provider_data['phone_number'],
'provider_photo_url': provider_data['photo_url'],
'provider_uid': provider_data['uid'],
'last_sign_in_timestamp': provider_data['last_sign_in_timestamp'],
}
writer.writerow(user_row)

#Upload to S3 bucket
s3 = boto3.client('s3', aws_access_key_id=aws_access_key_id, aws_secret_access_key=aws_secret_access_key)
s3.upload_file(csv_filename, bucket_name, 'firebase/auth/firebase_auth_from_airflow.csv')
print(f"File uploaded successfully to qa firebase/auth/firebase_auth_from_airflow.csv")

Continue reading...