[Python] Why is my Azure Functions deploy not working through GitLab CI/CD despite working...

I've created some Azure Functions using Python (3.9) and I'm having trouble setting up the CI/CD in the GitLab repository where the code resides. This repository resides in an Azure VM, not sure if that's relevant. The functions connect to an Azure PostgreSQL database, as you will see further down in the code.

The .gitlab-ci.yml triggers 3 jobs: The 2 first ones build and deploy a React app to Azure Static Web Apps successfully. The third one runs successfully in the GitLab pipelines but afterwards i see no functions in the Azure Function App overview page. I also don't see any error logs in the job.

When using the VSCode extension Azure Functions Core Tools to deploy directly to the Azure Function App, the deploy is successful and the functions show in the overview page, and they work correctly, connecting to the database and everything. I'm also able to run and test the functions locally.

Here are the different pieces of code involved. Some information will be ommited for privacy, like URLs

.gitlab-ci.yml:

stages:
- build-react
- deploy-react
- deploy-functions

# Build the React app
build-react:
stage: build-react
image: registry.gitlab.com/static-web-apps/azure-static-web-apps-deploy
script:
- cd app-frontend
- npm install
- VITE_FUNCTIONS_BASE_URL=azurewebsites-url npx vite build
artifacts:
paths:
- app-frontend/dist
only:
- main

# Deploy the React app to Azure Static Web Apps
deploy-react:
stage: deploy-react
image: registry.gitlab.com/static-web-apps/azure-static-web-apps-deploy
script:
- cd app-frontend
- npm install -g @azure/static-web-apps-cli
- swa deploy --env production --deployment-token token --output-location ./dist
only:
- main

deploy-functions:
stage: deploy-functions
image: mcr.microsoft.com/azure-functions/python:3.0 # Azure Functions image
script:
- set -e
- cd azure_functions # Navigate to the folder containing your Azure Function app
- zip -r ../my-functions.zip . # Create a zip file of only the contents of the azure_functions folder
- cd .. # Navigate back to the root directory
- echo "Zipped successfully"
- apt-get update && apt-get install -y ca-certificates curl apt-transport-https lsb-release gnupg
- curl -sL https://aka.ms/InstallAzureCLIDeb | sudo bash
- echo "Deploying to Azure..."
- az login --service-principal -u $AZURE_CLIENT_ID -p $AZURE_CLIENT_SECRET --tenant $AZURE_TENANT_ID
- az functionapp deployment source config-zip --resource-group $AZURE_RESOURCE_GROUP --name $AZURE_FUNCTIONAPP_NAME --src ./my-functions.zip
only:
- main



requirements.txt:

azure-functions
psycopg2-binary


function_app.py:

import json
import azure.functions as func
import logging
from datetime import datetime
import psycopg2
import os

# Database connection details
DB_HOST = os.getenv("DB_HOST")
DB_NAME = os.getenv("DB_NAME")
DB_USER = os.getenv("DB_USER")
DB_PASSWORD = os.getenv("DB_PASSWORD")
DB_PORT = os.getenv("DB_PORT")

app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)

@app.route(route="desks", methods=["GET"], auth_level=func.AuthLevel.ANONYMOUS)
def getDesks(req: func.HttpRequest) -> func.HttpResponse:
# Connect to database and fetch reservations for the given date
try:
connection = psycopg2.connect(
host=DB_HOST,
port=DB_PORT,
database=DB_NAME,
user=DB_USER,
password=DB_PASSWORD,
)

cursor = connection.cursor()

# Query to get all desks
query = """
SELECT d.id
FROM desks d;
"""

cursor.execute(query)
desks = cursor.fetchall()

# Prepare the response data
if desks:
response_data = [{"id": desk[0]} for desk in desks]
return func.HttpResponse(
body=json.dumps(response_data),
status_code=200,
mimetype="application/json",
)
else:
return func.HttpResponse("No desks found", status_code=200)

except Exception as e:
logging.error(f"Error fetching desks: {str(e)}")
return func.HttpResponse(
"An error occurred while processing the request.", status_code=500
)
finally:
if cursor:
cursor.close()
if connection:
connection.close()


@app.route(route="reservations", methods=["GET"], auth_level=func.AuthLevel.ANONYMOUS)
def getReservationsForDate(req: func.HttpRequest) -> func.HttpResponse:
# Get the date from the query parameter (expected format: YYYY-MM-DD)
reservation_date = req.params.get("date")
if not reservation_date:
return func.HttpResponse(
"Please provide a 'date' query parameter (YYYY-MM-DD).", status_code=400
)

try:
# Convert string date to a date object for validation
reservation_date = datetime.strptime(reservation_date, "%Y-%m-%d").date()
except ValueError:
return func.HttpResponse(
"Invalid date format. Please provide a valid date in YYYY-MM-DD format.",
status_code=400,
)

# Connect to database and fetch reservations for the given date
try:
connection = psycopg2.connect(
host=DB_HOST,
port=DB_PORT,
database=DB_NAME,
user=DB_USER,
password=DB_PASSWORD,
)

cursor = connection.cursor()

# Query to get all reservations for the provided date
query = """
SELECT r.id, r.desk_id, r.reserved_by, r.reservation_date
FROM reservations r
WHERE r.reservation_date = %s;
"""

cursor.execute(query, (reservation_date,))
reservations = cursor.fetchall()

# Prepare the response data
if reservations:
response_data = [
{
"id": res[0],
"desk_id": res[1],
"reserved_by": res[2],
"reservation_date": res[3].strftime("%Y-%m-%d"),
}
for res in reservations
]
return func.HttpResponse(
body=json.dumps(response_data),
status_code=200,
mimetype="application/json",
)
else:
return func.HttpResponse(body=json.dumps([]), status_code=200)

except Exception as e:
logging.error(f"Error fetching reservations: {str(e)}")
return func.HttpResponse(
"An error occurred while processing the request.", status_code=500
)
finally:
if cursor:
cursor.close()
if connection:
connection.close()


@app.route(
route="reservation-dates", methods=["GET"], auth_level=func.AuthLevel.ANONYMOUS
)
def getDatesWithReservationForEmail(req: func.HttpRequest) -> func.HttpResponse:
email = req.params.get("email")
if not email:
return func.HttpResponse(
"Please provide an 'email' query parameter.", status_code=400
)

todays_date = datetime.today().date()

# Connect to database and fetch dates in which the email provided has reservations
try:
connection = psycopg2.connect(
host=DB_HOST,
port=DB_PORT,
database=DB_NAME,
user=DB_USER,
password=DB_PASSWORD,
)

cursor = connection.cursor()

# Query to get all reservations for the provided date
query = """
SELECT r.reservation_date
FROM reservations r
WHERE r.reserved_by = %s AND r.reservation_date >= %s;
"""

cursor.execute(query, (email, todays_date))
reservation_dates = cursor.fetchall()

# Prepare the response data
if reservation_dates:
response_data = [res[0].strftime("%Y-%m-%d") for res in reservation_dates]
return func.HttpResponse(
body=json.dumps(response_data),
status_code=200,
mimetype="application/json",
)
else:
return func.HttpResponse(body=json.dumps([]), status_code=200)

except Exception as e:
logging.error(f"Error fetching reservation dates for email {email}: {str(e)}")
return func.HttpResponse(
"An error occurred while processing the request.", status_code=500
)
finally:
if cursor:
cursor.close()
if connection:
connection.close()


@app.route(route="reservations", methods=["POST"], auth_level=func.AuthLevel.ANONYMOUS)
def reserveDesks(req: func.HttpRequest) -> func.HttpResponse:
# Parse the JSON body
request_body = req.get_json()
if not request_body:
return func.HttpResponse("Invalid or missing JSON body.", status_code=400)

# Extract `userEmail`
employee_email = request_body.get("employee")
# Extract `reservations`
reservations = request_body.get("reservations", [])

# Validate presence of required fields
if not employee_email or not reservations:
return func.HttpResponse(
"Missing 'employee' or 'reservations' in the JSON body.",
status_code=400,
)

# Database connection
try:
connection = psycopg2.connect(
host=DB_HOST,
port=DB_PORT,
database=DB_NAME,
user=DB_USER,
password=DB_PASSWORD,
)

cursor = connection.cursor()
except Exception as e:
logging.error(f"Error connecting to the database: {str(e)}")
return func.HttpResponse("Database connection failed.", status_code=500)

# Connect to database and fetch reservations for the given date
try:
for reservation in reservations:
desk_id = reservation.get("desk_id")
reservation_date = reservation.get("reservation_date")

if not desk_id or not reservation_date:
return func.HttpResponse(
"Each reservation must include 'deskId' and 'reservationDate'.",
status_code=400,
)

checkIfSameReservationAlreadyExistsQuery = """
SELECT r.id, r.desk_id, r.reserved_by, r.reservation_date
FROM reservations r
WHERE r.reservation_date = %s AND r.desk_id = %s;
"""
cursor.execute(
checkIfSameReservationAlreadyExistsQuery, (reservation_date, desk_id)
)
reservations = cursor.fetchall()
if reservations and len(reservations) > 0:
response_data = [
{
"id": res[0],
"desk_id": res[1],
"reserved_by": res[2],
"reservation_date": res[3].strftime("%Y-%m-%d"),
}
for res in reservations
]
return func.HttpResponse(
body=json.dumps(response_data),
status_code=409,
mimetype="application/json",
)

# Insert reservation into the database
insert_query = """
INSERT INTO reservations (desk_id, reserved_by, reservation_date)
VALUES (%s, %s, %s)
"""
cursor.execute(insert_query, (desk_id, employee_email, reservation_date))

# Commit the transaction
connection.commit()

except Exception as e:
logging.error(f"Error fetching reservations: {str(e)}")
return func.HttpResponse(
"An error occurred while processing the request.", status_code=500
)
finally:
if cursor:
cursor.close()
if connection:
connection.close()

# Return success response
return func.HttpResponse(
f"Successfully created {len(reservations)} reservations for {employee_email}.",
status_code=201,
)


@app.route(
route="reservations", methods=["DELETE"], auth_level=func.AuthLevel.ANONYMOUS
)
def cancelReservations(req: func.HttpRequest) -> func.HttpResponse:
# Get reservation_ids from query parameter
reservation_ids_str = req.params.get("reservation_ids")
if not reservation_ids_str:
return func.HttpResponse(
"Please provide reservation_ids as a query parameter.", status_code=400
)

try:
# Split the reservation_ids and convert them into a list of integers
reservation_ids = list(map(int, reservation_ids_str.split(",")))

# Connect to database and delete the reservations
connection = psycopg2.connect(
host=DB_HOST,
port=DB_PORT,
database=DB_NAME,
user=DB_USER,
password=DB_PASSWORD,
)

cursor = connection.cursor()

placeholders = ", ".join("%s" for _ in reservation_ids)
# Query to delete the reservations by IDs
query = f"""
DELETE FROM reservations
WHERE id IN ({placeholders});
"""
cursor.execute(query, reservation_ids)
connection.commit()

# Check how many rows were deleted
deleted_rows = cursor.rowcount

if deleted_rows > 0:
return func.HttpResponse(
f"Successfully deleted {deleted_rows} reservation(s).", status_code=200
)
else:
return func.HttpResponse(
"No reservations found with the provided IDs.", status_code=404
)

except Exception as e:
logging.error(f"Error deleting reservations: {str(e)}")
return func.HttpResponse(
"An error occurred while processing the request.", status_code=500
)

finally:
if cursor:
cursor.close()
if connection:
connection.close()

Continue reading...