<p dir="auto"><strong>Apache Airflow version</strong>:<br>
2.0.0</p>
<p dir="auto"><strong>Kubernetes version (if you are using kubernetes)</strong> (use <code class="notranslate">kubectl version</code>):<br>
1.14</p>
<p dir="auto"><strong>Environment</strong>:</p>
<ul dir="auto">
<li><strong>Cloud provider or hardware configuration</strong>:</li>
<li><strong>OS</strong> (e.g. from /etc/os-release):</li>
<li><strong>Kernel</strong> (e.g. <code class="notranslate">uname -a</code>):</li>
<li><strong>Install tools</strong>:</li>
<li><strong>Others</strong>:</li>
</ul>
<p dir="auto"><strong>What happened</strong>:</p>
<p dir="auto">[2021-04-02 07:23:30,513] [ERROR] app.py:1892 - Exception on /api/v1/dags/auto_test/dagRuns [POST]<br>
Traceback (most recent call last):<br>
File "/usr/local/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1277, in _execute_context<br>
cursor, statement, parameters, context<br>
File "/usr/local/lib/python3.6/site-packages/sqlalchemy/engine/default.py", line 608, in do_execute<br>
cursor.execute(statement, parameters)<br>
psycopg2.errors.UniqueViolation: duplicate key value violates unique constraint "dag_run_dag_id_execution_date_key"<br>
DETAIL:  Key (dag_id, execution_date)=(auto_test, 1967-12-13 20:57:42.043+01) already exists.</p>
<p dir="auto">The above exception was the direct cause of the following exception:</p>
<p dir="auto">Traceback (most recent call last):<br>
File "/usr/local/lib/python3.6/site-packages/flask/app.py", line 2447, in wsgi_app<br>
response = self.full_dispatch_request()<br>
File "/usr/local/lib/python3.6/site-packages/flask/app.py", line 1952, in full_dispatch_request<br>
rv = self.handle_user_exception(e)<br>
File "/usr/local/lib/python3.6/site-packages/flask/app.py", line 1821, in handle_user_exception<br>
reraise(exc_type, exc_value, tb)<br>
File "/usr/local/lib/python3.6/site-packages/flask/_compat.py", line 39, in reraise<br>
raise value<br>
File "/usr/local/lib/python3.6/site-packages/flask/app.py", line 1950, in full_dispatch_request<br>
rv = self.dispatch_request()<br>
File "/usr/local/lib/python3.6/site-packages/flask/app.py", line 1936, in dispatch_request<br>
return self.view_functions<a href="**req.view_args">rule.endpoint</a><br>
File "/usr/local/lib/python3.6/site-packages/connexion/decorators/decorator.py", line 48, in wrapper<br>
response = function(request)<br>
File "/usr/local/lib/python3.6/site-packages/connexion/decorators/uri_parsing.py", line 144, in wrapper<br>
response = function(request)<br>
File "/usr/local/lib/python3.6/site-packages/connexion/decorators/validation.py", line 184, in wrapper<br>
response = function(request)<br>
File "/usr/local/lib/python3.6/site-packages/connexion/decorators/validation.py", line 384, in wrapper<br>
return function(request)<br>
File "/usr/local/lib/python3.6/site-packages/connexion/decorators/response.py", line 103, in wrapper<br>
response = function(request)<br>
File "/usr/local/lib/python3.6/site-packages/connexion/decorators/parameter.py", line 121, in wrapper<br>
return function(**kwargs)<br>
File "/usr/local/lib/python3.6/site-packages/airflow/api_connexion/security.py", line 47, in decorated<br>
return func(*args, **kwargs)<br>
File "/usr/local/lib/python3.6/site-packages/airflow/utils/session.py", line 65, in wrapper<br>
return func(*args, session=session, **kwargs)<br>
File "/usr/local/lib/python3.6/site-packages/airflow/api_connexion/endpoints/dag_run_endpoint.py", line 231, in post_dag_run<br>
session.commit()<br>
File "/usr/local/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 1046, in commit<br>
self.transaction.commit()<br>
File "/usr/local/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 504, in commit<br>
self._prepare_impl()<br>
File "/usr/local/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 483, in _prepare_impl<br>
self.session.flush()<br>
File "/usr/local/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 2540, in flush<br>
self._flush(objects)<br>
File "/usr/local/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 2682, in _flush<br>
transaction.rollback(<em>capture_exception=True)<br>
File "/usr/local/lib/python3.6/site-packages/sqlalchemy/util/langhelpers.py", line 70, in <strong>exit</strong><br>
with_traceback=exc_tb,<br>
File "/usr/local/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 182, in raise</em><br>
raise exception<br>
File "/usr/local/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 2642, in _flush<br>
flush_context.execute()<br>
File "/usr/local/lib/python3.6/site-packages/sqlalchemy/orm/unitofwork.py", line 422, in execute<br>
rec.execute(self)<br>
File "/usr/local/lib/python3.6/site-packages/sqlalchemy/orm/unitofwork.py", line 589, in execute<br>
uow,<br>
File "/usr/local/lib/python3.6/site-packages/sqlalchemy/orm/persistence.py", line 245, in save_obj<br>
insert,<br>
File "/usr/local/lib/python3.6/site-packages/sqlalchemy/orm/persistence.py", line 1136, in _emit_insert_statements<br>
statement, params<br>
File "/usr/local/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1011, in execute<br>
return meth(self, multiparams, params)<br>
File "/usr/local/lib/python3.6/site-packages/sqlalchemy/sql/elements.py", line 298, in _execute_on_connection<br>
return connection._execute_clauseelement(self, multiparams, params)<br>
File "/usr/local/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1130, in _execute_clauseelement<br>
distilled_params,<br>
File "/usr/local/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1317, in <em>execute_context<br>
e, statement, parameters, cursor, context<br>
File "/usr/local/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1511, in <em>handle_dbapi_exception<br>
sqlalchemy_exception, with_traceback=exc_info[2], from</em>=e<br>
File "/usr/local/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 182, in raise</em><br>
raise exception<br>
File "/usr/local/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1277, in _execute_context<br>
cursor, statement, parameters, context<br>
File "/usr/local/lib/python3.6/site-packages/sqlalchemy/engine/default.py", line 608, in do_execute<br>
cursor.execute(statement, parameters)<br>
sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "dag_run_dag_id_execution_date_key"<br>
DETAIL:  Key (dag_id, execution_date)=(auto_test, 1967-12-13 20:57:42.043+01) already exists.</p>
<p dir="auto">[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]<br>
[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': &lt;DagRunType.MANUAL: 'manual'&gt;, 'conf': &lt;psycopg2.extensions.Binary object at 0x7f07b30b71e8&gt;, 'last_scheduling_decision': None, 'dag_hash': None}]</p>

<p dir="auto"><strong>What you expected to happen</strong>:<br>
The second trigger success</p>

<p dir="auto"><strong>How to reproduce it</strong>:<br>
Trigger a dag with the following conf:</p>
<p dir="auto"><code class="notranslate">{ "dag_run_id": "dag_run_id_zstp_4435_postman", "execution_date": "1967-12-13T19:57:42.043Z", "conf": {} }</code></p>
<p dir="auto">Then trigerring the same dag only with a different dag_run_id:</p>
<p dir="auto"><code class="notranslate">{ "dag_run_id": "dag_run_id_zstp_4435_postman111", "execution_date": "1967-12-13T19:57:42.043Z", "conf": {} }</code></p>

<p dir="auto"><strong>Anything else we need to know</strong>:</p>
<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.5.2</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">NOTE: I am building application based on ReactFlow and Backend in Python using Airflow</p>
<p dir="auto">Please review this issue on stackoverflow.</p>
<p dir="auto"><a href="https://stackoverflow.com/questions/41560614/airflow-this-dag-isnt-available-in-the-webserver-dagbag-object" rel="nofollow">Airflow "This DAG isnt available in the webserver DagBag object "</a></p>
<p dir="auto">Problem:</p>
<p dir="auto">I have following code which dynamically creates Dag file and store it in /Users/[username]/airflow</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="#
# 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 &gt;&gt; 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&quot;DAG with id {__id} is present in the DagBag.&quot;)
    else:
        # Print statement that dag is not present.
        print(f&quot;DAG with id {__id} is not present in the DagBag.&quot;)


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&quot;{dagId}.py&quot;

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

    # Write DAG to file.
    with open(os.path.join(os.environ['AIRFLOW_HOME'], 'dags', fileName), &quot;w&quot;, 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"><pre class="notranslate"><code class="notranslate">#
# 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 &gt;&gt; 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
</code></pre></div>
<p dir="auto">Now when i Run <code class="notranslate">CreateDag</code> it creates file in dags folder and works like charm.</p>
<p dir="auto">Screenshot of Dags Folder</p>
<p dir="auto"><a target="_blank" rel="noopener noreferrer nofollow" href="https://user-images.githubusercontent.com/76559517/225761133-e09b1497-7a77-423a-b622-f5faa3800938.png"><img width="918" alt="Screenshot 2023-03-17 at 3 25 49 AM" src="https://user-images.githubusercontent.com/76559517/225761133-e09b1497-7a77-423a-b622-f5faa3800938.png" style="max-width: 100%;"></a></p>
<p dir="auto">and Dag do appear in Airflow webApp screenshot is below.</p>
<p dir="auto"><a target="_blank" rel="noopener noreferrer nofollow" href="https://user-images.githubusercontent.com/76559517/225761257-4d72228d-e0d9-4788-a28f-126c1f1d794c.png"><img width="1728" alt="Screenshot 2023-03-17 at 3 26 36 AM" src="https://user-images.githubusercontent.com/76559517/225761257-4d72228d-e0d9-4788-a28f-126c1f1d794c.png" style="max-width: 100%;"></a></p>
<p dir="auto">Problem is that when i click on Dag it throws error that <strong>DAG "Amazinsg" seems to be missing from DagBag.</strong> screenshot is below</p>
<p dir="auto"><a target="_blank" rel="noopener noreferrer nofollow" href="https://user-images.githubusercontent.com/76559517/225761416-498edcee-d9ba-4288-a18e-c4dd82d104c4.png"><img width="1728" alt="Screenshot 2023-03-17 at 3 27 35 AM" src="https://user-images.githubusercontent.com/76559517/225761416-498edcee-d9ba-4288-a18e-c4dd82d104c4.png" style="max-width: 100%;"></a></p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto"><a target="_blank" rel="noopener noreferrer nofollow" href="https://user-images.githubusercontent.com/76559517/225762828-bc67c845-60bc-4a55-bb88-2628a74a25da.png"><img width="747" alt="Screenshot 2023-03-17 at 3 32 59 AM" src="https://user-images.githubusercontent.com/76559517/225762828-bc67c845-60bc-4a55-bb88-2628a74a25da.png" style="max-width: 100%;"></a></p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">OS Ventura</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">apache-airflow-providers-celery==3.1.0<br>
apache-airflow-providers-common-sql==1.3.4<br>
apache-airflow-providers-ftp==3.3.1<br>
apache-airflow-providers-http==4.2.0<br>
apache-airflow-providers-imap==3.1.1<br>
apache-airflow-providers-sqlite==3.3.1</p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Virtualenv installation</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Anything else</h3>
<p dir="auto"><g-emoji class="g-emoji" alias="broken_heart" fallback-src="https://github.githubassets.com/images/icons/emoji/unicode/1f494.png">💔</g-emoji> 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.</p>
<p dir="auto">Stackoverflow: <a href="Stackoverflow">https://stackoverflow.com/questions/41560614/airflow-this-dag-isnt-available-in-the-webserver-dagbag-object</a></p>
<p dir="auto">Regards</p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>