Issue 1:
Apache Airflow version:

2.0.0

Kubernetes version (if you are using kubernetes) (use kubectl version):

1.14

Environment:

Cloud provider or hardware configuration:

OS (e.g. from /etc/os-release):

Kernel (e.g. uname -a):

Install tools:

Others:

What happened:

[2021-04-02 07:23:30,513] [ERROR] app.py:1892 - Exception on /api/v1/dags/auto_test/dagRuns [POST]

Traceback (most recent call last):

File "/usr/local/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1277, in _execute_context

cursor, statement, parameters, context

File "/usr/local/lib/python3.6/site-packages/sqlalchemy/engine/default.py", line 608, in do_execute

cursor.execute(statement, parameters)

psycopg2.errors.UniqueViolation: duplicate key value violates unique constraint "dag_run_dag_id_execution_date_key"

DETAIL: Key (dag_id, execution_date)=(auto_test, 1967-12-13 20:57:42.043+01) already exists.

The above exception was the direct cause of the following exception:

Traceback (most recent call last):

File "/usr/local/lib/python3.6/site-packages/flask/app.py", line 2447, in wsgi_app

response = self.full_dispatch_request()

File "/usr/local/lib/python3.6/site-packages/flask/app.py", line 1952, in full_dispatch_request

rv = self.handle_user_exception(e)

File "/usr/local/lib/python3.6/site-packages/flask/app.py", line 1821, in handle_user_exception

reraise(exc_type, exc_value, tb)

File "/usr/local/lib/python3.6/site-packages/flask/_compat.py", line 39, in reraise

raise value

File "/usr/local/lib/python3.6/site-packages/flask/app.py", line 1950, in full_dispatch_request

rv = self.dispatch_request()

File "/usr/local/lib/python3.6/site-packages/flask/app.py", line 1936, in dispatch_request

return self.view_functionsrule.endpoint

File "/usr/local/lib/python3.6/site-packages/connexion/decorators/decorator.py", line 48, in wrapper

response = function(request)

File "/usr/local/lib/python3.6/site-packages/connexion/decorators/uri_parsing.py", line 144, in wrapper

response = function(request)

File "/usr/local/lib/python3.6/site-packages/connexion/decorators/validation.py", line 184, in wrapper

response = function(request)

File "/usr/local/lib/python3.6/site-packages/connexion/decorators/validation.py", line 384, in wrapper

return function(request)

File "/usr/local/lib/python3.6/site-packages/connexion/decorators/response.py", line 103, in wrapper

response = function(request)

File "/usr/local/lib/python3.6/site-packages/connexion/decorators/parameter.py", line 121, in wrapper

return function(**kwargs)

File "/usr/local/lib/python3.6/site-packages/airflow/api_connexion/security.py", line 47, in decorated

return func(*args, **kwargs)

File "/usr/local/lib/python3.6/site-packages/airflow/utils/session.py", line 65, in wrapper

return func(*args, session=session, **kwargs)

File "/usr/local/lib/python3.6/site-packages/airflow/api_connexion/endpoints/dag_run_endpoint.py", line 231, in post_dag_run

session.commit()

File "/usr/local/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 1046, in commit

self.transaction.commit()

File "/usr/local/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 504, in commit

self._prepare_impl()

File "/usr/local/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 483, in _prepare_impl

self.session.flush()

File "/usr/local/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 2540, in flush

self._flush(objects)

File "/usr/local/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 2682, in _flush

transaction.rollback(capture_exception=True)

File "/usr/local/lib/python3.6/site-packages/sqlalchemy/util/langhelpers.py", line 70, in exit

with_traceback=exc_tb,

File "/usr/local/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 182, in raise

raise exception

File "/usr/local/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 2642, in _flush

flush_context.execute()

File "/usr/local/lib/python3.6/site-packages/sqlalchemy/orm/unitofwork.py", line 422, in execute

rec.execute(self)

File "/usr/local/lib/python3.6/site-packages/sqlalchemy/orm/unitofwork.py", line 589, in execute

uow,

File "/usr/local/lib/python3.6/site-packages/sqlalchemy/orm/persistence.py", line 245, in save_obj

insert,

File "/usr/local/lib/python3.6/site-packages/sqlalchemy/orm/persistence.py", line 1136, in _emit_insert_statements

statement, params

File "/usr/local/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1011, in execute

return meth(self, multiparams, params)

File "/usr/local/lib/python3.6/site-packages/sqlalchemy/sql/elements.py", line 298, in _execute_on_connection

return connection._execute_clauseelement(self, multiparams, params)

File "/usr/local/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1130, in _execute_clauseelement

distilled_params,

File "/usr/local/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1317, in execute_context

e, statement, parameters, cursor, context

File "/usr/local/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1511, in handle_dbapi_exception

sqlalchemy_exception, with_traceback=exc_info[2], from=e

File "/usr/local/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 182, in raise

raise exception

File "/usr/local/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1277, in _execute_context

cursor, statement, parameters, context

File "/usr/local/lib/python3.6/site-packages/sqlalchemy/engine/default.py", line 608, in do_execute

cursor.execute(statement, parameters)

sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "dag_run_dag_id_execution_date_key"

DETAIL: Key (dag_id, execution_date)=(auto_test, 1967-12-13 20:57:42.043+01) already exists.

[SQL: INSERT INTO dag_run (dag_id, execution_date, start_date, end_date, state, run_id, creating_job_id, external_trigger, run_type, conf, last_scheduling_decision, dag_hash) VALUES (%(dag_id)s, %(execution_date)s, %(start_date)s, %(end_date)s, %(state)s, %(run_id)s, %(creating_job_id)s, %(external_trigger)s, %(run_type)s, %(conf)s, %(last_scheduling_decision)s, %(dag_hash)s) RETURNING dag_run.id]

[parameters: {'dag_id': 'auto_test', 'execution_date': datetime.datetime(1967, 12, 13, 19, 57, 42, 43000, tzinfo=Timezone('UTC')), 'start_date': datetime.datetime(2021, 4, 2, 7, 23, 30, 511735, tzinfo=Timezone('UTC')), 'end_date': None, 'state': 'running', 'run_id': 'dag_run_id_zstp_4435_postman11', 'creating_job_id': None, 'external_trigger': True, 'run_type': <DagRunType.MANUAL: 'manual'>, 'conf': <psycopg2.extensions.Binary object at 0x7f07b30b71e8>, 'last_scheduling_decision': None, 'dag_hash': None}]

What you expected to happen:

The second trigger success

How to reproduce it:

Trigger a dag with the following conf:

{ "dag_run_id": "dag_run_id_zstp_4435_postman", "execution_date": "1967-12-13T19:57:42.043Z", "conf": {} }

Then trigerring the same dag only with a different dag_run_id:

{ "dag_run_id": "dag_run_id_zstp_4435_postman111", "execution_date": "1967-12-13T19:57:42.043Z", "conf": {} }

Anything else we need to know:

Issue 2:
Apache Airflow version

2.5.2

What happened

NOTE: I am building application based on ReactFlow and Backend in Python using Airflow

Please review this issue on stackoverflow.

Airflow "This DAG isnt available in the webserver DagBag object "

Problem:

I have following code which dynamically creates Dag file and store it in /Users/[username]/airflow

#
# IMPORTS
#
import os # CORE: Operating System module for file system operations.
import inspect # CORE: Inspect module for getting source code of a function.
from datetime import datetime # CORE: Datetime module for date and time operations.
from airflow.models import DAG, DagBag # PIP: Airflow DAG and DagBag modules for DAG operations.
from airflow.operators.python_operator import PythonOperator # PIP: Airflow PythonOperator module for Python operations.

#
# ENVS
#
os.environ['KMP_DUPLICATE_LIB_OK'] = 'True'

#
# FUNCTIONS
#
def DagFunctionToExecute():
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash_operator import BashOperator

default_args = {
'owner': 'airflow',
'depends_on_past': False,
'start_date': datetime(2023, 3, 16),
'email': ['airflow@example.com'],
'email_on_failure': False,
'email_on_retry': False,
'retries': 1,
'retry_delay': timedelta(minutes=5),
}

dag = DAG('Amazing', default_args=default_args, schedule_interval=timedelta(days=1))

t1 = BashOperator(
task_id='print_date',
bash_command='date',
dag=dag)

t2 = BashOperator(
task_id='sleep',
bash_command='sleep 5',
retries=3,
dag=dag)

t1 >> t2

def CheckDagInDagBag(__id):
# Load all tags from dagbag folder.
dagbag = DagBag(dag_folder=os.path.join(os.environ['AIRFLOW_HOME'], 'dags'))

print(dagbag.dags, os.path.join(os.environ['AIRFLOW_HOME'], 'dags'))

# Check if DAG is present in the DagBag.
if dagbag.dags.get(__id):
# Print statement that dag is present.
print(f"DAG with id {__id} is present in the DagBag.")
else:
# Print statement that dag is not present.
print(f"DAG with id {__id} is not present in the DagBag.")

def CreateOperators(__dag, __edges, __functionToExecute):
# Variable assignment.
operators = {}

# Loop over edges and convert them to operators.
for item in __edges:
source = item['source']
target = item['target']

# Check if source and target are not in operators.
if source not in operators:
# Update operator with source and python operator
# with callable python function.
operators[source] = PythonOperator(
task_id=source,
python_callable=__functionToExecute,
provide_context=True,
dag=__dag,
op_kwargs={'source': source, '__dagId': __dag.dag_id}
)

# Check if target is not in operators.
if target not in operators:
# Update operator with target and python operator
operators[target] = PythonOperator(
task_id=target,
python_callable=__functionToExecute,
provide_context=True,
dag=__dag,
op_kwargs={'source': source, '__dagId': __dag.dag_id}
)

# Set upstream of target to source.
operators[target].set_upstream(operators[source])

# Return operators.
return operators

def SyncDagToDb(__dag):
# Sync DAG to database.
__dag.sync_to_db()

# Check if DAG is present in the DagBag.
CheckDagInDagBag(__dag.dag_id)

def WriteDagToFile(__dag, __functionToExecute):
# Variable assignment.
dagId = __dag.dag_id
fileName = f"{dagId}.py"

# Get source code of DAG.
sourceCode = inspect.getsource(__functionToExecute)

# Write DAG to file.
with open(os.path.join(os.environ['AIRFLOW_HOME'], 'dags', fileName), "w", encoding='utf8') as __dagFile:
# Write source to the file.
__dagFile.write(sourceCode)

def CreateDag(__edges, __title, __description):
# Create DAG.
dag = DAG(
dag_id=__title,
description=__description,
start_date=datetime.now(),
default_view='tree',
schedule_interval='@daily',
catchup=False
)

# Create operators and sync DAG to database.
# along side to it write dag to file.
CreateOperators(dag, __edges, DagFunctionToExecute)
SyncDagToDb(dag)
WriteDagToFile(dag, DagFunctionToExecute)

# Return dag.
return dag

Now when i Run CreateDag it creates file in dags folder and works like charm.

Screenshot of Dags Folder

and Dag do appear in Airflow webApp screenshot is below.

Problem is that when i click on Dag it throws error that DAG "Amazinsg" seems to be missing from DagBag. screenshot is below

What you think should happen instead

No response

How to reproduce

Operating System

OS Ventura

Versions of Apache Airflow Providers

apache-airflow-providers-celery==3.1.0

apache-airflow-providers-common-sql==1.3.4

apache-airflow-providers-ftp==3.3.1

apache-airflow-providers-http==4.2.0

apache-airflow-providers-imap==3.1.1

apache-airflow-providers-sqlite==3.3.1

Deployment

Virtualenv installation

Deployment details

No response

Anything else

💔 This issue is opened from last 6 Years on Stackoverflow and still it is unresolved. its a request to you guys please help me out in this i dont move away from airflow because i have already spent last 5 Days coding whole server around it. Hope you guys understand.

Stackoverflow: https://stackoverflow.com/questions/41560614/airflow-this-dag-isnt-available-in-the-webserver-dagbag-object

Regards

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct