<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">Other Airflow 2 version (please specify below)</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">Version Airflow: 2.5.1</p>
<p dir="auto">I have a file in python that generate dynamic DAG, and all time when have a new code in that file, is necessary to execute <code class="notranslate">./airflow.sh dags reserialize</code>, but sometimes this command return with an error like below, but :</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="[2023-04-13 19:46:54,669] {dag.py:2690} INFO - Sync 35 DAGs
Traceback (most recent call last):
  File &quot;/home/airflow/.local/bin/airflow&quot;, line 8, in &lt;module&gt;
    sys.exit(main())
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/__main__.py&quot;, line 39, in main
    args.func(args)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/cli/cli_parser.py&quot;, line 52, in command
    return func(*args, **kwargs)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/session.py&quot;, line 75, in wrapper
    return func(*args, session=session, **kwargs)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/cli.py&quot;, line 108, in wrapper
    return f(*args, **kwargs)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/cli/commands/dag_command.py&quot;, line 492, in dag_reserialize
    dagbag.sync_to_db(session=session)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/session.py&quot;, line 72, in wrapper
    return func(*args, **kwargs)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/models/dagbag.py&quot;, line 645, in sync_to_db
    for attempt in run_with_db_retries(logger=self.log):
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/tenacity/__init__.py&quot;, line 384, in __iter__
    do = self.iter(retry_state=retry_state)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/tenacity/__init__.py&quot;, line 351, in iter
    return fut.result()
  File &quot;/usr/local/lib/python3.7/concurrent/futures/_base.py&quot;, line 428, in result
    return self.__get_result()
  File &quot;/usr/local/lib/python3.7/concurrent/futures/_base.py&quot;, line 384, in __get_result
    raise self._exception
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/models/dagbag.py&quot;, line 660, in sync_to_db
    self.dags.values(), processor_subdir=processor_subdir, session=session
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/session.py&quot;, line 72, in wrapper
    return func(*args, **kwargs)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/models/dag.py&quot;, line 2701, in bulk_write_to_db
    orm_dags: list[DagModel] = with_row_locks(query, of=DagModel, session=session).all()
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/query.py&quot;, line 2773, in all
    return self._iter().all()
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/query.py&quot;, line 2919, in _iter
    execution_options={&quot;_sa_orm_load_options&quot;: self.load_options},
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/session.py&quot;, line 1713, in execute
    conn = self._connection_for_bind(bind)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/session.py&quot;, line 1553, in _connection_for_bind              2-tetris-linkedin-hourly-days', 'fileloc': '/opt/airflow/dags/register_dynamic_dag.py', 'fileloc_hash': 51105027638417678, 'data': '{&quot;__version&quot;: 1, &quot;dag&quot;: {&quot;_dag_id&quot;: &quot;96ddcc3b-900a-44a7-bda9-81b9eeserialized DAG: /opt/airflow/dags/register_alerts_resport_dag.py
    engine, execution_options       python3.7/site-packages/airflow/models/dagbag.py&quot;, line 631, in _serialize_dag_capturing_errors
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/session.py&quot;, line 721, in _connection_for_bind
    self._assert_active()
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/session.py&quot;, line 608, in _assert_active
    code=&quot;7s2a&quot;,
sqlalchemy.exc.PendingRollbackError: This Session's transaction has been rolled back due to a previous exception during flush. To begin a new transaction with this Session, first issue Session.rollback(). Original exception was: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint &quot;serialized_dag_pkey&quot;
DETAIL:  Key (dag_id)=(96ddcc3b-900a-44a7-bda9-81b9eefde4d2-dynamic-dag-hourly-days) already exists."><pre class="notranslate"><code class="notranslate">[2023-04-13 19:46:54,669] {dag.py:2690} INFO - Sync 35 DAGs
Traceback (most recent call last):
  File "/home/airflow/.local/bin/airflow", line 8, in &lt;module&gt;
    sys.exit(main())
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/__main__.py", line 39, in main
    args.func(args)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/cli/cli_parser.py", line 52, in command
    return func(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/session.py", line 75, in wrapper
    return func(*args, session=session, **kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/cli.py", line 108, in wrapper
    return f(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/cli/commands/dag_command.py", line 492, in dag_reserialize
    dagbag.sync_to_db(session=session)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/session.py", line 72, in wrapper
    return func(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/dagbag.py", line 645, in sync_to_db
    for attempt in run_with_db_retries(logger=self.log):
  File "/home/airflow/.local/lib/python3.7/site-packages/tenacity/__init__.py", line 384, in __iter__
    do = self.iter(retry_state=retry_state)
  File "/home/airflow/.local/lib/python3.7/site-packages/tenacity/__init__.py", line 351, in iter
    return fut.result()
  File "/usr/local/lib/python3.7/concurrent/futures/_base.py", line 428, in result
    return self.__get_result()
  File "/usr/local/lib/python3.7/concurrent/futures/_base.py", line 384, in __get_result
    raise self._exception
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/dagbag.py", line 660, in sync_to_db
    self.dags.values(), processor_subdir=processor_subdir, session=session
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/session.py", line 72, in wrapper
    return func(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/dag.py", line 2701, in bulk_write_to_db
    orm_dags: list[DagModel] = with_row_locks(query, of=DagModel, session=session).all()
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/query.py", line 2773, in all
    return self._iter().all()
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/query.py", line 2919, in _iter
    execution_options={"_sa_orm_load_options": self.load_options},
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/session.py", line 1713, in execute
    conn = self._connection_for_bind(bind)
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/session.py", line 1553, in _connection_for_bind              2-tetris-linkedin-hourly-days', 'fileloc': '/opt/airflow/dags/register_dynamic_dag.py', 'fileloc_hash': 51105027638417678, 'data': '{"__version": 1, "dag": {"_dag_id": "96ddcc3b-900a-44a7-bda9-81b9eeserialized DAG: /opt/airflow/dags/register_alerts_resport_dag.py
    engine, execution_options       python3.7/site-packages/airflow/models/dagbag.py", line 631, in _serialize_dag_capturing_errors
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/session.py", line 721, in _connection_for_bind
    self._assert_active()
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/session.py", line 608, in _assert_active
    code="7s2a",
sqlalchemy.exc.PendingRollbackError: This Session's transaction has been rolled back due to a previous exception during flush. To begin a new transaction with this Session, first issue Session.rollback(). Original exception was: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "serialized_dag_pkey"
DETAIL:  Key (dag_id)=(96ddcc3b-900a-44a7-bda9-81b9eefde4d2-dynamic-dag-hourly-days) already exists.
</code></pre></div>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">I think that airflow should only reserialize without problem with duplicate key.</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">The python's file that register dynamic dag (I shortened code to focus on the problem):</p>
<p dir="auto">Here, the data is fixed as dict on python, but in real scene is a json file, that update sometime for other DAG.</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="import pytz
from datetime import datetime, timedelta
from airflow.decorators import dag, task

data = {
    'companies': [{
        'id': '1233',
        'slug': 'company-1',
        'timezone': {
            'id': 'America/Sao_Paulo'
        },
        'connections': [{
            'id': 'facebook',
        }, {
            'id': 'instagram',
        }]
    }, {
        'id': '1234',
        'slug': 'company-2',
        'timezone': {
            'id': 'America/Sao_Paulo'
        },
        'connections': [{
            'id': 'facebook',   
        }, {
            'id': 'instagram',
        }, {
            'id': 'twitter',
        }]
    }]
}

def create_dynamic_dag(dag_id, company):
    schedule_interval = &quot;* 5 * * * *&quot;
    local_tz = pytz.timezone(company['timezone']['id'])
    start_date = datetime.now(local_tz) - timedelta(days=7, hours=0)

    @dag(
        dag_id=dag_id,
        schedule=schedule_interval,
        start_date=start_date,
        catchup=False,
        max_active_runs=1,
    )
    def dynamic_dag():
        @task.branch
        def ping():
            # ...code to ping
            # return 'done' or 'notify_error'
            pass

        @task
        def notify_error():
            # ...code to notify error
            pass

        @task
        def done():
            # ... code to done
            pass
        
        ping_instance = ping()

        notify_error_instance = notify_error()
        done_instance = done()

        ping_instance &gt;&gt; [done_instance, notify_error_instance]
    
    dagD = dynamic_dag()

    globals()[dag_id] = dagD

    return dagD

for company in data['companies']:
    for connection in company['connections']:
        dag_id = f&quot;{company['id']}-{company['slug']}-{connection['id']}&quot;
        dagD = create_dynamic_dag(dag_id, company)"><pre class="notranslate"><code class="notranslate">import pytz
from datetime import datetime, timedelta
from airflow.decorators import dag, task

data = {
    'companies': [{
        'id': '1233',
        'slug': 'company-1',
        'timezone': {
            'id': 'America/Sao_Paulo'
        },
        'connections': [{
            'id': 'facebook',
        }, {
            'id': 'instagram',
        }]
    }, {
        'id': '1234',
        'slug': 'company-2',
        'timezone': {
            'id': 'America/Sao_Paulo'
        },
        'connections': [{
            'id': 'facebook',   
        }, {
            'id': 'instagram',
        }, {
            'id': 'twitter',
        }]
    }]
}

def create_dynamic_dag(dag_id, company):
    schedule_interval = "* 5 * * * *"
    local_tz = pytz.timezone(company['timezone']['id'])
    start_date = datetime.now(local_tz) - timedelta(days=7, hours=0)

    @dag(
        dag_id=dag_id,
        schedule=schedule_interval,
        start_date=start_date,
        catchup=False,
        max_active_runs=1,
    )
    def dynamic_dag():
        @task.branch
        def ping():
            # ...code to ping
            # return 'done' or 'notify_error'
            pass

        @task
        def notify_error():
            # ...code to notify error
            pass

        @task
        def done():
            # ... code to done
            pass
        
        ping_instance = ping()

        notify_error_instance = notify_error()
        done_instance = done()

        ping_instance &gt;&gt; [done_instance, notify_error_instance]
    
    dagD = dynamic_dag()

    globals()[dag_id] = dagD

    return dagD

for company in data['companies']:
    for connection in company['connections']:
        dag_id = f"{company['id']}-{company['slug']}-{connection['id']}"
        dagD = create_dynamic_dag(dag_id, company)
</code></pre></div>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Mac, but I'm using docker from airflow</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Docker-Compose</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">I only follow the documentation of <a href="https://airflow.apache.org/docs/apache-airflow/stable/howto/docker-compose/index.html" rel="nofollow">Airflow in docker</a>.</p>
<h3 dir="auto">Anything else</h3>
<p dir="auto">Sometime this errors doesn't happen, but it's happens with more frequency.</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"> 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>
<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.3.3 (latest released)</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">When manually triggering a dag run using "Trigger DAG w/ config" and setting the logical date and time, the UI of the dag run triggered will show a data interval that is two schedule intervals off. For example, for a daily run, setting the logical date to 2022-07-24 00:00:00 UTC would show the data interval with a start date of 2022-07-22 and end of 2022-07-23. Rendering {{ ds_nodash }} on this run will show "2022-07-24" which is correct, but since the grid view orders the runs on the incorrect start date, things can get confusing.</p>
<p dir="auto"><a target="_blank" rel="noopener noreferrer nofollow" href="https://user-images.githubusercontent.com/46539900/181820332-0eedab1d-f163-46f0-8fa7-9f6c1d24d8b8.png"><img width="648" alt="image" src="https://user-images.githubusercontent.com/46539900/181820332-0eedab1d-f163-46f0-8fa7-9f6c1d24d8b8.png" style="max-width: 100%;"></a></p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">Either the data interval behaves the same as with the scheduled run and show start date as the logical date and the end date as the logical date + schedule interval, or, per the documentation <a href="https://airflow.apache.org/docs/apache-airflow/stable/faq.html#why-next-ds-or-prev-ds-might-not-contain-expected-values" rel="nofollow">here</a> the prev_ds == next_ds == ds and so the data interval should all be 2022-07-24 for the above example.</p>
<h3 dir="auto">How to reproduce</h3>
<ol dir="auto">
<li>Create a simple DAG with a schedule interval, as such:</li>
</ol>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="import json
from datetime import datetime

from airflow.decorators import dag, task


@dag(
    schedule_interval=&quot;@daily&quot;,
    start_date=datetime(2021, 1, 1),
    catchup=False,
)
def example_dag_basic():
    @task()
    def example():
        print(&quot;this is an example task&quot;)

    example()


example_dag_basic = example_dag_basic()"><pre class="notranslate"><code class="notranslate">import json
from datetime import datetime

from airflow.decorators import dag, task


@dag(
    schedule_interval="@daily",
    start_date=datetime(2021, 1, 1),
    catchup=False,
)
def example_dag_basic():
    @task()
    def example():
        print("this is an example task")

    example()


example_dag_basic = example_dag_basic()
</code></pre></div>
<ol start="2" dir="auto">
<li>Trigger a run using "Trigger DAG w/ config" and manually create a run for some date</li>
<li>Observe the start date as 2 schedules off.</li>
</ol>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Debian Bullseye</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Astronomer</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">Local was tested on docker compose (from astro-cli)</p>
<h3 dir="auto">Anything else</h3>
<p dir="auto"><em>No response</em></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"> 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>