You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "Steexyz (via GitHub)" <gi...@apache.org> on 2026/01/14 17:04:26 UTC
[I] The conn_id `xxxxxxx` isn't defined Google Cloud Platform [airflow]
Steexyz opened a new issue, #60535:
URL: https://github.com/apache/airflow/issues/60535
### Apache Airflow Provider(s)
google
### Versions of Apache Airflow Providers
19.3.0
### Apache Airflow version
3.1.6
### Operating System
Debian GNU/Linux 12 (bookworm)
### Deployment
Docker-Compose
### Deployment details
Deployment was bade using the base docker-compose file.
```
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
#
# Basic Airflow cluster configuration for CeleryExecutor with Redis and PostgreSQL.
#
# WARNING: This configuration is for local development. Do not use it in a production deployment.
#
# This configuration supports basic configuration using environment variables or an .env file
# The following variables are supported:
#
# AIRFLOW_IMAGE_NAME - Docker image name used to run Airflow.
# Default: apache/airflow:3.1.6
# AIRFLOW_UID - User ID in Airflow containers
# Default: 50000
# AIRFLOW_PROJ_DIR - Base path to which all the files will be volumed.
# Default: .
# Those configurations are useful mostly in case of standalone testing/running Airflow in test/try-out mode
#
# _AIRFLOW_WWW_USER_USERNAME - Username for the administrator account (if requested).
# Default: airflow
# _AIRFLOW_WWW_USER_PASSWORD - Password for the administrator account (if requested).
# Default: airflow
# _PIP_ADDITIONAL_REQUIREMENTS - Additional PIP requirements to add when starting all containers.
# Use this option ONLY for quick checks. Installing requirements at container
# startup is done EVERY TIME the service is started.
# A better way is to build a custom image or extend the official image
# as described in https://airflow.apache.org/docs/docker-stack/build.html.
# Default: ''
#
# Feel free to modify this file to suit your needs.
---
x-airflow-common:
&airflow-common
# In order to add custom dependencies or upgrade provider distributions you can use your extended image.
# Comment the image line, place your Dockerfile in the directory where you placed the docker-compose.yaml
# and uncomment the "build" line below, Then run `docker-compose build` to build the images.
image: custom-airflow-3:3.1.6
# build: .
environment:
&airflow-common-env
AIRFLOW__CORE__EXECUTOR: CeleryExecutor
AIRFLOW__CORE__AUTH_MANAGER: airflow.providers.fab.auth_manager.fab_auth_manager.FabAuthManager
AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
AIRFLOW__CORE__FERNET_KEY: '8_4t1Ld5YV7sS3k9n-Bv2pXgZc1R0fHwJmNlUa6I7oQ='
AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: 'true'
AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql://airflow:airflow@postgres/airflow
AIRFLOW__CELERY__BROKER_URL: redis://:@redis:6379/0
AIRFLOW__CORE__LOAD_EXAMPLES: 'false'
AIRFLOW__CORE__EXECUTION_API_SERVER_URL: 'http://airflow-apiserver:8080/execution/'
# yamllint disable rule:line-length
# Use simple http server on scheduler for health checks
# See https://airflow.apache.org/docs/apache-airflow/stable/administration-and-deployment/logging-monitoring/check-health.html#scheduler-health-check-server
# yamllint enable rule:line-length
AIRFLOW__SCHEDULER__ENABLE_HEALTH_CHECK: 'true'
# WARNING: Use _PIP_ADDITIONAL_REQUIREMENTS option ONLY for a quick checks
# The following line can be used to set a custom config file, stored in the local config folder
AIRFLOW_CONFIG: '/opt/airflow/config/airflow.cfg'
volumes:
- ./dags:/opt/airflow/dags
- ./logs:/opt/airflow/logs
- ./plugins:/opt/airflow/plugins
- ./config:/opt/airflow/config
- ${appdata}/gcloud:/home/airflow/.config/gcloud
user: "${AIRFLOW_UID:-50000}:0"
depends_on:
&airflow-common-depends-on
redis:
condition: service_healthy
postgres:
condition: service_healthy
services:
postgres:
image: postgres:16
environment:
POSTGRES_USER: airflow
POSTGRES_PASSWORD: airflow
POSTGRES_DB: airflow
volumes:
- postgres-db-volume:/var/lib/postgresql/data
healthcheck:
test: ["CMD", "pg_isready", "-U", "airflow"]
interval: 10s
retries: 5
start_period: 5s
restart: always
redis:
# Redis is limited to 7.2-bookworm due to licencing change
# https://redis.io/blog/redis-adopts-dual-source-available-licensing/
image: redis:7.2-bookworm
expose:
- 6379
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 10s
timeout: 30s
retries: 50
start_period: 30s
restart: always
airflow-apiserver:
<<: *airflow-common
command: api-server
ports:
- "8080:8080"
healthcheck:
test: ["CMD", "curl", "--fail", "http://localhost:8080/api/v2/version"]
interval: 30s
timeout: 10s
retries: 5
start_period: 30s
restart: always
depends_on:
<<: *airflow-common-depends-on
airflow-init:
condition: service_completed_successfully
airflow-scheduler:
<<: *airflow-common
command: scheduler
healthcheck:
test: ["CMD", "curl", "--fail", "http://localhost:8974/health"]
interval: 30s
timeout: 10s
retries: 5
start_period: 30s
restart: always
depends_on:
<<: *airflow-common-depends-on
airflow-init:
condition: service_completed_successfully
airflow-dag-processor:
<<: *airflow-common
command: dag-processor
healthcheck:
test: ["CMD-SHELL", 'airflow jobs check --job-type DagProcessorJob --hostname "$${HOSTNAME}"']
interval: 30s
timeout: 10s
retries: 5
start_period: 30s
restart: always
depends_on:
<<: *airflow-common-depends-on
airflow-init:
condition: service_completed_successfully
airflow-triggerer:
<<: *airflow-common
command: triggerer
healthcheck:
test: ["CMD-SHELL", 'airflow jobs check --job-type TriggererJob --hostname "$${HOSTNAME}"']
interval: 30s
timeout: 10s
retries: 5
start_period: 30s
restart: always
depends_on:
<<: *airflow-common-depends-on
airflow-init:
condition: service_completed_successfully
airflow-worker:
<<: *airflow-common
command: celery worker
healthcheck:
# yamllint disable rule:line-length
test:
- "CMD-SHELL"
- 'celery --app airflow.providers.celery.executors.celery_executor.app inspect ping -d "celery@$${HOSTNAME}" || celery --app airflow.executors.celery_executor.app inspect ping -d "celery@$${HOSTNAME}"'
interval: 30s
timeout: 10s
retries: 5
start_period: 30s
environment:
<<: *airflow-common-env
# Required to handle warm shutdown of the celery workers properly
# See https://airflow.apache.org/docs/docker-stack/entrypoint.html#signal-propagation
DUMB_INIT_SETSID: "0"
restart: always
depends_on:
<<: *airflow-common-depends-on
airflow-apiserver:
condition: service_healthy
airflow-init:
condition: service_completed_successfully
airflow-init:
<<: *airflow-common
entrypoint: /bin/bash
# yamllint disable rule:line-length
command:
- -c
- |
if [[ -z "50000" ]]; then
echo
echo -e "\033[1;33mWARNING!!!: AIRFLOW_UID not set!\e[0m"
echo "If you are on Linux, you SHOULD follow the instructions below to set "
echo "AIRFLOW_UID environment variable, otherwise files will be owned by root."
echo "For other operating systems you can get rid of the warning with manually created .env file:"
echo " See: https://airflow.apache.org/docs/apache-airflow/stable/howto/docker-compose/index.html#setting-the-right-airflow-user"
echo
export AIRFLOW_UID=$$(id -u)
fi
one_meg=1048576
mem_available=$$(($$(getconf _PHYS_PAGES) * $$(getconf PAGE_SIZE) / one_meg))
cpus_available=$$(grep -cE 'cpu[0-9]+' /proc/stat)
disk_available=$$(df / | tail -1 | awk '{print $$4}')
warning_resources="false"
if (( mem_available < 4000 )) ; then
echo
echo -e "\033[1;33mWARNING!!!: Not enough memory available for Docker.\e[0m"
echo "At least 4GB of memory required. You have $$(numfmt --to iec $$((mem_available * one_meg)))"
echo
warning_resources="true"
fi
if (( cpus_available < 2 )); then
echo
echo -e "\033[1;33mWARNING!!!: Not enough CPUS available for Docker.\e[0m"
echo "At least 2 CPUs recommended. You have $${cpus_available}"
echo
warning_resources="true"
fi
if (( disk_available < one_meg * 10 )); then
echo
echo -e "\033[1;33mWARNING!!!: Not enough Disk space available for Docker.\e[0m"
echo "At least 10 GBs recommended. You have $$(numfmt --to iec $$((disk_available * 1024 )))"
echo
warning_resources="true"
fi
if [[ $${warning_resources} == "true" ]]; then
echo
echo -e "\033[1;33mWARNING!!!: You have not enough resources to run Airflow (see above)!\e[0m"
echo "Please follow the instructions to increase amount of resources available:"
echo " https://airflow.apache.org/docs/apache-airflow/stable/howto/docker-compose/index.html#before-you-begin"
echo
fi
echo
echo "Creating missing opt dirs if missing:"
echo
mkdir -v -p /opt/airflow/{logs,dags,plugins,config}
echo
echo "Airflow version:"
/entrypoint airflow version
echo
echo "Files in shared volumes:"
echo
ls -la /opt/airflow/{logs,dags,plugins,config}
echo
echo "Running airflow config list to create default config file if missing."
echo
/entrypoint airflow config list >/dev/null
echo
echo "Files in shared volumes:"
echo
ls -la /opt/airflow/{logs,dags,plugins,config}
echo
echo "Change ownership of files in /opt/airflow to 50000:0"
echo
chown -R "50000:0" /opt/airflow/
echo
echo "Change ownership of files in shared volumes to 50000:0"
echo
chown -v -R "50000:0" /opt/airflow/{logs,dags,plugins,config}
echo
echo "Files in shared volumes:"
echo
ls -la /opt/airflow/{logs,dags,plugins,config}
# yamllint enable rule:line-length
environment:
<<: *airflow-common-env
_AIRFLOW_DB_MIGRATE: 'true'
_AIRFLOW_WWW_USER_CREATE: 'true'
_AIRFLOW_WWW_USER_USERNAME: ${_AIRFLOW_WWW_USER_USERNAME:-airflow}
_AIRFLOW_WWW_USER_PASSWORD: ${_AIRFLOW_WWW_USER_PASSWORD:-airflow}
_PIP_ADDITIONAL_REQUIREMENTS: ''
user: "0:0"
airflow-cli:
<<: *airflow-common
profiles:
- debug
environment:
<<: *airflow-common-env
CONNECTION_CHECK_MAX_COUNT: "0"
# Workaround for entrypoint issue. See: https://github.com/apache/airflow/issues/16252
command:
- bash
- -c
- airflow
depends_on:
<<: *airflow-common-depends-on
volumes:
postgres-db-volume:
```
A custom image has been made with additionnal requirements
```
apache-airflow-providers-fab
apache-airflow-providers-celery
connexion[swagger-ui]
apache-airflow-providers-jdbc
apache-airflow-providers-oracle[common.sql]
apache-airflow-providers-snowflake
apache-airflow-providers-docker
airflow-provider-rabbitmq
apache-airflow-providers-google
google-cloud-storage
pandas
zeep
xmltodict
orjson
pyodbc
```
Docker file
```dockerfile
FROM apache/airflow:3.1.6-python3.12
ARG AIRFLOW_VERSION=3.1.6
ARG PYTHON_VERSION=3.12
ARG CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt"
# Switch to the root user to install system-level dependencies
USER root
# Install system dependencies, including OpenJDK for JDBC
RUN apt-get update && \
apt-get install -y --no-install-recommends curl gnupg telnet openjdk-17-jdk && \
curl https://packages.microsoft.com/keys/microsoft.asc | gpg --dearmor > /etc/apt/trusted.gpg.d/microsoft.gpg && \
curl https://packages.microsoft.com/config/debian/12/prod.list > /etc/apt/sources.list.d/mssql-release.list && \
apt-get update && \
ACCEPT_EULA=Y apt-get install -y --no-install-recommends msodbcsql18 mssql-tools18 unixodbc-dev && \
apt-get clean && \
rm -rf /var/lib/apt/lists/*
# Set JAVA_HOME environment variable
ENV JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64
# Set the container's timezone to UTC
ENV TZ=UTC
ENV PATH="$PATH:/opt/mssql-tools18/bin"
# Allow legacy TLS versions in Java's security configuration
RUN sed -i 's/TLSv1, TLSv1.1, //g' /usr/lib/jvm/java-17-openjdk-amd64/conf/security/java.security
# Create directory for JDBC drivers and download the driver
RUN mkdir -p /opt/airflow/drivers && \
curl -o /opt/airflow/drivers/mssql-jdbc.jar https://repo1.maven.org/maven2/com/microsoft/sqlserver/mssql-jdbc/12.4.2.jre11/mssql-jdbc-12.4.2.jre11.jar && \
curl -o /opt/airflow/drivers/ojdbc8-12.2.0.1.jar https://repo1.maven.org/maven2/com/oracle/database/jdbc/ojdbc8/12.2.0.1/ojdbc8-12.2.0.1.jar
COPY --chown=airflow:airflow openedge.jar /opt/airflow/drivers/openedge.jar
# Switch back to the non-privileged airflow user
USER airflow
# Install Python packages
COPY requirements.txt .
RUN pip install -r requirements.txt \
--constraint "${CONSTRAINT_URL}"
```
### What happened
I am not able to retrieve a Google Cloud connection anymore.
```python
from airflow.providers.google.common.hooks.base_google import GoogleBaseHook
```
```python
hook = GoogleBaseHook(gcp_conn_id=self.connection_id)
```
Produce the following error:
```
Fichier "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", ligne 1004 dans run
Fichier "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", ligne 1405 dans _execute_task
Fichier "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/bases/operator.py", ligne 417 dans wrapper
Fichier "/opt/airflow/plugins/operators/extraction.py", ligne 100 dans execute
Fichier "/opt/airflow/plugins/operators/extraction.py", ligne 87 dans execute
Fichier "/opt/airflow/plugins/connectors/cloud/target/gcs_target_connector.py", ligne 35 dans __enter__
Fichier "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/google/common/hooks/base_google.py", ligne 283 dans __init__
Fichier "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/bases/hook.py", ligne 61 dans get_connection
Fichier "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/definitions/connection.py", ligne 225 dans get
Fichier "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/context.py", ligne 174 dans _get_connection
```
The connection can successfully be retrieved from CLI using the following command:
```
airflow connections list
10 | gcs-bucket-project | google_cloud_platform | None | None | None | None | None | None | False | False | {'project': | google-cloud-platform
```
### What you think should happen instead
I should be able to retrieve the connection like previously in Airflow 2.11.0. I haven't seen any documentation that would prove otherwise but I'm may have not stumbled upon it yet.
### How to reproduce
Try to instantiate a hook with a Google Platform connection using Application Default Credential (ADC)
### Anything else
This may or may not be an issue and be related to changes in Airflow 3 but my JDBC connections are not suffering from the same issue using airflow.hooks.base.BaseHook. I've also tried using BaseHook to retry the connections informations without success.
### Are you willing to submit PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] The conn_id `xxxxxxx` isn't defined Google Cloud Platform [airflow]
Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
shahar1 commented on issue #60535:
URL: https://github.com/apache/airflow/issues/60535#issuecomment-3801383485
> Ok, some more insights. I was able to get it working.
>
> See gcs_target_connector.py, most of the changes were to moved the Hook initialisation out of the **enter** method.
>
> from google.cloud import storage
> from airflow.exceptions import AirflowException
> from connectors.base_connector import TargetConnector
> from airflow.providers.google.cloud.hooks.gcs import GCSHook
> from airflow.providers.google.common.hooks.base_google import GoogleBaseHook
>
> class GCSTargetConnector(TargetConnector):
> """
> Target connector for writing data to Google Cloud Storage (GCS).
> Handles file naming, bucket/project resolution, and client setup.
> """
> def __init__(self, conn_id="google_cloud_default", bucket_name=None,
> project_id=None, gcs_path=None, table_name=None, file_prefix="stream"):
> """
> Initialize the GCS target connector.
> conn_id: Airflow connection ID for GCS.
> bucket_name: Name of the GCS bucket to write to.
> project_id: GCP project ID.
> gcs_path: Path in the bucket to write files.
> table_name: Optional table name for metadata.
> file_prefix: Prefix for output files.
> """
> super().__init__(conn_id)
> self.bucket_name = bucket_name
> self.project_id = project_id
> self.gcs_path = gcs_path
> self.table_name = table_name
> self.file_prefix = file_prefix
> self._part_num = 0
> self._hook = None
>
> @property
> def hook(self):
> if self._hook is None:
> # The Task SDK will resolve this only when first accessed
> self.log.info(f"Resolving connection: {self.connection_id}")
> self._hook = GoogleBaseHook(gcp_conn_id=self.connection_id)
> return self._hook
>
> def __enter__(self):
> # 1. Initialize the hook.
> # Note: 'gcp_conn_id' is the standard param for GoogleCloudBaseHook/GoogleBaseHook
> # self.log.info("Before getting hook")
> # hook = GoogleBaseHook(gcp_conn_id=self.connection_id)
> # self.log.info("After getting hook")
> # # 2. Explicitly fetch credentials.
> # # This is where ADC is triggered.
> # credentials = hook.get_credentials()
> # self.log.info("After getting credentials")
>
> # # 3. Resolve Project ID.
> # # For ADC, 'Project Id' in the Airflow Connection UI is still highly recommended.
> # self._effective_project_id = self.project_id or hook.project_id
>
> # if not self._effective_project_id:
> # raise AirflowException("GCP Project ID could not be determined from Connection or ADC.")
>
> # self._effective_bucket_name = self.bucket_name or self._effective_project_id
>
> # self.log.info("Before initializing client")
>
> # # 4. Initialize storage client
> # self.client = storage.Client(
> # project=self._effective_project_id,
> # credentials=credentials
> # )
> # self.log.info("After initializing client")
> self.log.info("Entering GCS context...")
> return self
>
> def save(self, log, data: bytes, context: dict):
> if not data:
> return None
>
> # Use the property 'self.hook' to ensure _hook is initialized
> hook_obj = self.hook
>
> # Initialize client once or check if it exists
> if self.client is None:
> self.log.info("Initializing GCS Storage Client for the first batch...")
> credentials = hook_obj.get_credentials()
> self._effective_project_id = self.project_id or hook_obj.project_id
>
> if not self._effective_project_id:
> raise AirflowException("GCP Project ID could not be determined.")
>
> self._effective_bucket_name = self.bucket_name or self._effective_project_id
> self.client = storage.Client(
> project=self._effective_project_id,
> credentials=credentials
> )
>
> logical_date = context['logical_date']
> formatted_date = logical_date.strftime("%Y%m%d%H%M%S%f")[:-3]
>
> file_name = f"{self.gcs_path}/{self.table_name}/{self.file_prefix}_{self.table_name}_{formatted_date}_{self._part_num}.json"
>
> bucket = self.client.bucket(self._effective_bucket_name)
> blob = bucket.blob(file_name)
>
> log.info(f"Uploading to gs://{self._effective_bucket_name}/{file_name}")
> blob.upload_from_string(data, content_type="application/jsonl")
>
> self._part_num += 1
> return f"gs://{self._effective_bucket_name}/{file_name}"
> According to Gemini:
>
> 1. Connection Retrieval: The "Thin Worker" Shift
> In Airflow 2, your worker was "thick"—it had a direct connection to the metadata database. In Airflow 3, workers are "thin" and isolated for security and scalability. They communicate with an Internal API through the Task SDK.
>
> The Eager Hook Problem: The GoogleBaseHook source code shows that it attempts to fetch connection details immediately upon instantiation to populate its "extras".
>
> Initialization Timing: When you instantiated this in **enter**, the Task SDK hadn't yet fully "hydrated" the task environment. The worker tried to reach the API before the proxy was ready, resulting in the AirflowNotFoundException.
>
> The Lazy Solution: By moving the hook to a [@Property](https://github.com/Property), you deferred the connection lookup until the save method was called. At that point, the task is in its "active" phase, and the Task SDK is fully synchronized with the Internal API, allowing the connection to be resolved instantly.
>
> Not sure if that makes any sense or if there's an issue with the task SDK.
It does make sense for me that it's related to an expected behavior of the Task SDK in Airflow 3 - please read the AIP for more details.
As of Airflow 3, you cannot initialize connections directly in the DAG’s code (as they don't have access to the meta database to fetch connection details), and it is done only during their execution with the Task SDK interface.
I'm converting this issue to a discussion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] The conn_id `xxxxxxx` isn't defined Google Cloud Platform [airflow]
Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
shahar1 closed issue #60535: The conn_id `xxxxxxx` isn't defined Google Cloud Platform
URL: https://github.com/apache/airflow/issues/60535
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] The conn_id `xxxxxxx` isn't defined Google Cloud Platform [airflow]
Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
shahar1 commented on issue #60535:
URL: https://github.com/apache/airflow/issues/60535#issuecomment-3795645662
Could you please provide a full Dag example?
It seems to be related to the changes introduced by [AIP-72](https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-72+Task+Execution+Interface+aka+Task+SDK) (Task SDK), which prevent direct access from the Dags to the database on running.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] The conn_id `xxxxxxx` isn't defined Google Cloud Platform [airflow]
Posted by "Steexyz (via GitHub)" <gi...@apache.org>.
Steexyz commented on issue #60535:
URL: https://github.com/apache/airflow/issues/60535#issuecomment-3800779814
Ok, some more insights. I was able to get it working.
See gcs_target_connector.py, most of the changes were to moved the Hook initialisation out of the __enter__ method.
```python
from google.cloud import storage
from airflow.exceptions import AirflowException
from connectors.base_connector import TargetConnector
from airflow.providers.google.cloud.hooks.gcs import GCSHook
from airflow.providers.google.common.hooks.base_google import GoogleBaseHook
class GCSTargetConnector(TargetConnector):
"""
Target connector for writing data to Google Cloud Storage (GCS).
Handles file naming, bucket/project resolution, and client setup.
"""
def __init__(self, conn_id="google_cloud_default", bucket_name=None,
project_id=None, gcs_path=None, table_name=None, file_prefix="stream"):
"""
Initialize the GCS target connector.
conn_id: Airflow connection ID for GCS.
bucket_name: Name of the GCS bucket to write to.
project_id: GCP project ID.
gcs_path: Path in the bucket to write files.
table_name: Optional table name for metadata.
file_prefix: Prefix for output files.
"""
super().__init__(conn_id)
self.bucket_name = bucket_name
self.project_id = project_id
self.gcs_path = gcs_path
self.table_name = table_name
self.file_prefix = file_prefix
self._part_num = 0
self._hook = None
@property
def hook(self):
if self._hook is None:
# The Task SDK will resolve this only when first accessed
self.log.info(f"Resolving connection: {self.connection_id}")
self._hook = GoogleBaseHook(gcp_conn_id=self.connection_id)
return self._hook
def __enter__(self):
# 1. Initialize the hook.
# Note: 'gcp_conn_id' is the standard param for GoogleCloudBaseHook/GoogleBaseHook
# self.log.info("Before getting hook")
# hook = GoogleBaseHook(gcp_conn_id=self.connection_id)
# self.log.info("After getting hook")
# # 2. Explicitly fetch credentials.
# # This is where ADC is triggered.
# credentials = hook.get_credentials()
# self.log.info("After getting credentials")
# # 3. Resolve Project ID.
# # For ADC, 'Project Id' in the Airflow Connection UI is still highly recommended.
# self._effective_project_id = self.project_id or hook.project_id
# if not self._effective_project_id:
# raise AirflowException("GCP Project ID could not be determined from Connection or ADC.")
# self._effective_bucket_name = self.bucket_name or self._effective_project_id
# self.log.info("Before initializing client")
# # 4. Initialize storage client
# self.client = storage.Client(
# project=self._effective_project_id,
# credentials=credentials
# )
# self.log.info("After initializing client")
self.log.info("Entering GCS context...")
return self
def save(self, log, data: bytes, context: dict):
if not data:
return None
# Use the property 'self.hook' to ensure _hook is initialized
hook_obj = self.hook
# Initialize client once or check if it exists
if self.client is None:
self.log.info("Initializing GCS Storage Client for the first batch...")
credentials = hook_obj.get_credentials()
self._effective_project_id = self.project_id or hook_obj.project_id
if not self._effective_project_id:
raise AirflowException("GCP Project ID could not be determined.")
self._effective_bucket_name = self.bucket_name or self._effective_project_id
self.client = storage.Client(
project=self._effective_project_id,
credentials=credentials
)
logical_date = context['logical_date']
formatted_date = logical_date.strftime("%Y%m%d%H%M%S%f")[:-3]
file_name = f"{self.gcs_path}/{self.table_name}/{self.file_prefix}_{self.table_name}_{formatted_date}_{self._part_num}.json"
bucket = self.client.bucket(self._effective_bucket_name)
blob = bucket.blob(file_name)
log.info(f"Uploading to gs://{self._effective_bucket_name}/{file_name}")
blob.upload_from_string(data, content_type="application/jsonl")
self._part_num += 1
return f"gs://{self._effective_bucket_name}/{file_name}"
```
According to Gemini:
---------------
1. Connection Retrieval: The "Thin Worker" Shift
In Airflow 2, your worker was "thick"—it had a direct connection to the metadata database. In Airflow 3, workers are "thin" and isolated for security and scalability. They communicate with an Internal API through the Task SDK.
The Eager Hook Problem: The GoogleBaseHook source code shows that it attempts to fetch connection details immediately upon instantiation to populate its "extras".
Initialization Timing: When you instantiated this in __enter__, the Task SDK hadn't yet fully "hydrated" the task environment. The worker tried to reach the API before the proxy was ready, resulting in the AirflowNotFoundException.
The Lazy Solution: By moving the hook to a @property, you deferred the connection lookup until the save method was called. At that point, the task is in its "active" phase, and the Task SDK is fully synchronized with the Internal API, allowing the connection to be resolved instantly.
---------------
Not sure if that makes any sense or if there's an issue with the task SDK.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] The conn_id `xxxxxxx` isn't defined Google Cloud Platform [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on issue #60535:
URL: https://github.com/apache/airflow/issues/60535#issuecomment-3750604470
Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] The conn_id `xxxxxxx` isn't defined Google Cloud Platform [airflow]
Posted by "Steexyz (via GitHub)" <gi...@apache.org>.
Steexyz commented on issue #60535:
URL: https://github.com/apache/airflow/issues/60535#issuecomment-3799796994
Hi @shahar1,
Here's a DAG example that used to work with 2.11.
```python
from __future__ import annotations
from airflow import DAG
import pendulum
from operators.extraction import ExtractionOperator
from connectors.jdbc.source.oracle_source_connector import OracleSourceConnector
from connectors.cloud.target.gcs_target_connector import GCSTargetConnector
with DAG(
dag_id='ar_collectors',
start_date=pendulum.datetime(2025, 10, 1, tz="America/Toronto"),
schedule="0 10 * * *",
catchup=False,
tags=['jdbc', 'oracleebs', 'ar_collectors']
) as dag:
# 1. Define the Source (Oracle)
oracle_source = OracleSourceConnector(
conn_id='oracleebs',
sql="SELECT * FROM AR.AR_COLLECTORS",
query_mode='delta',
delta_column='LAST_UPDATE_DATE',
xcom_key='ar_collectors_delta'
)
# 2. Define the Target (GCS)
gcs_target = GCSTargetConnector(
conn_id='gcs-bucket-project',
table_name='AR_COLLECTORS',
gcs_path='raw/OracleEBS'
)
# 3. Use the Orchestrator (ExtractionOperator)
ExtractionOperator(
task_id='select_ar_collectors',
source=oracle_source,
targets=[gcs_target]
)
```
FYI, the OracleSourceConnector does retrieve its connection and can retrieve the data.
The Oracle Source Connector retrieves its connection using this parent class:
```python
from airflow.hooks.base import BaseHook
from airflow.exceptions import AirflowException
class BaseHook(BaseHook):
"""Base class for all internal hooks to standardize logging and error handling."""
def __init__(self, conn_id: str, **kwargs):
super().__init__(**kwargs)
self.conn_id = conn_id
def get_connection_metadata(self):
"""Standardized connection retrieval with clear error messaging."""
try:
return self.get_connection(self.conn_id)
except Exception:
self.log.error(f"Failed to find Airflow Connection: {self.conn_id}")
raise AirflowException(f"Connection {self.conn_id} is missing or inaccessible.")
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org