Fix NULL constraints

Author: vincbeck

airflow-core/src/airflow/models/backfill.py

@@ -120,7 +120,7 @@ class Backfill(Base):
     from_date: Mapped[UtcDateTime] = mapped_column(UtcDateTime, nullable=False)
     to_date: Mapped[UtcDateTime] = mapped_column(UtcDateTime, nullable=False)
     dag_run_conf: Mapped[JSONField] = mapped_column(JSONField(json=json), nullable=False, default={})
-    is_paused: Mapped[bool] = mapped_column(Boolean, default=False)
+    is_paused: Mapped[bool | None] = mapped_column(Boolean, default=False, nullable=True)
     """
     Controls whether new dag runs will be created for this backfill.
 
@@ -172,7 +172,7 @@ class BackfillDagRun(Base):
     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
     backfill_id: Mapped[int] = mapped_column(Integer, nullable=False)
     dag_run_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
-    exception_reason: Mapped[str] = mapped_column(StringID())
+    exception_reason: Mapped[str | None] = mapped_column(StringID(), nullable=True)
     logical_date: Mapped[UtcDateTime] = mapped_column(UtcDateTime, nullable=False)
     sort_ordinal: Mapped[int] = mapped_column(Integer, nullable=False)
 

airflow-core/src/airflow/models/connection.py

@@ -130,18 +130,18 @@ class Connection(Base, LoggingMixin):
     id: Mapped[int] = mapped_column(Integer(), primary_key=True)
     conn_id: Mapped[str] = mapped_column(String(ID_LEN), unique=True, nullable=False)
     conn_type: Mapped[str] = mapped_column(String(500), nullable=False)
-    description: Mapped[str] = mapped_column(
-        Text().with_variant(Text(5000), "mysql").with_variant(String(5000), "sqlite")
+    description: Mapped[str | None] = mapped_column(
+        Text().with_variant(Text(5000), "mysql").with_variant(String(5000), "sqlite"), nullable=True
     )
-    host: Mapped[str] = mapped_column(String(500))
-    schema: Mapped[str] = mapped_column(String(500))
-    login: Mapped[str] = mapped_column(Text())
-    _password: Mapped[str] = mapped_column("password", Text())
-    port: Mapped[int] = mapped_column(Integer())
+    host: Mapped[str | None] = mapped_column(String(500), nullable=True)
+    schema: Mapped[str | None] = mapped_column(String(500), nullable=True)
+    login: Mapped[str | None] = mapped_column(Text(), nullable=True)
+    _password: Mapped[str | None] = mapped_column("password", Text(), nullable=True)
+    port: Mapped[int | None] = mapped_column(Integer(), nullable=True)
     is_encrypted: Mapped[bool] = mapped_column(Boolean, unique=False, default=False)
     is_extra_encrypted: Mapped[bool] = mapped_column(Boolean, unique=False, default=False)
     team_id: Mapped[str | None] = mapped_column(UUIDType(binary=False), ForeignKey("team.id"), nullable=True)
-    _extra: Mapped[str] = mapped_column("extra", Text())
+    _extra: Mapped[str | None] = mapped_column("extra", Text(), nullable=True)
 
     def __init__(
         self,

airflow-core/src/airflow/models/dag.py

@@ -322,29 +322,29 @@ class DagModel(Base):
     # Whether that DAG was seen on the last DagBag load
     is_stale: Mapped[bool] = mapped_column(Boolean, default=True)
     # Last time the scheduler started
-    last_parsed_time: Mapped[UtcDateTime] = mapped_column(UtcDateTime)
+    last_parsed_time: Mapped[UtcDateTime | None] = mapped_column(UtcDateTime, nullable=True)
     # How long it took to parse this file
-    last_parse_duration: Mapped[float] = mapped_column(Float)
+    last_parse_duration: Mapped[float | None] = mapped_column(Float, nullable=True)
     # Time when the DAG last received a refresh signal
     # (e.g. the DAG's "refresh" button was clicked in the web UI)
-    last_expired: Mapped[UtcDateTime] = mapped_column(UtcDateTime)
+    last_expired: Mapped[UtcDateTime | None] = mapped_column(UtcDateTime, nullable=True)
     # The location of the file containing the DAG object
     # Note: Do not depend on fileloc pointing to a file; in the case of a
     # packaged DAG, it will point to the subpath of the DAG within the
     # associated zip.
-    fileloc: Mapped[str] = mapped_column(String(2000))
-    relative_fileloc: Mapped[str] = mapped_column(String(2000))
+    fileloc: Mapped[str | None] = mapped_column(String(2000), nullable=True)
+    relative_fileloc: Mapped[str | None] = mapped_column(String(2000), nullable=True)
     bundle_name: Mapped[str] = mapped_column(StringID(), ForeignKey("dag_bundle.name"), nullable=False)
     # The version of the bundle the last time the DAG was processed
     bundle_version: Mapped[str | None] = mapped_column(String(200), nullable=True)
     # String representing the owners
-    owners: Mapped[str] = mapped_column(String(2000))
+    owners: Mapped[str | None] = mapped_column(String(2000), nullable=True)
     # Display name of the dag
     _dag_display_property_value: Mapped[str | None] = mapped_column(
         "dag_display_name", String(2000), nullable=True
     )
     # Description of the dag
-    description: Mapped[str] = mapped_column(Text)
+    description: Mapped[str | None] = mapped_column(Text, nullable=True)
     # Timetable summary
     timetable_summary: Mapped[str | None] = mapped_column(Text, nullable=True)
     # Timetable description
@@ -374,14 +374,14 @@ class DagModel(Base):
     has_import_errors: Mapped[bool] = mapped_column(Boolean(), default=False, server_default="0")
 
     # The logical date of the next dag run.
-    next_dagrun: Mapped[UtcDateTime] = mapped_column(UtcDateTime)
+    next_dagrun: Mapped[UtcDateTime | None] = mapped_column(UtcDateTime, nullable=True)
 
     # Must be either both NULL or both datetime.
-    next_dagrun_data_interval_start: Mapped[UtcDateTime] = mapped_column(UtcDateTime)
-    next_dagrun_data_interval_end: Mapped[UtcDateTime] = mapped_column(UtcDateTime)
+    next_dagrun_data_interval_start: Mapped[UtcDateTime | None] = mapped_column(UtcDateTime, nullable=True)
+    next_dagrun_data_interval_end: Mapped[UtcDateTime | None] = mapped_column(UtcDateTime, nullable=True)
 
     # Earliest time at which this ``next_dagrun`` can be created.
-    next_dagrun_create_after: Mapped[UtcDateTime] = mapped_column(UtcDateTime)
+    next_dagrun_create_after: Mapped[UtcDateTime | None] = mapped_column(UtcDateTime, nullable=True)
 
     __table_args__ = (Index("idx_next_dagrun_create_after", next_dagrun_create_after, unique=False),)
 

airflow-core/src/airflow/models/dag_favorite.py

@@ -17,16 +17,12 @@
 # under the License.
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
-
 from sqlalchemy import ForeignKey
+from sqlalchemy.orm import Mapped
 
 from airflow.models.base import Base, StringID
 from airflow.utils.sqlalchemy import mapped_column
 
-if TYPE_CHECKING:
-    from sqlalchemy.orm import Mapped
-
 
 class DagFavorite(Base):
     """Association table model linking users to their favorite DAGs."""

airflow-core/src/airflow/models/dag_version.py

@@ -49,7 +49,7 @@ class DagVersion(Base):
     )
     dag_model = relationship("DagModel", back_populates="dag_versions")
     bundle_name: Mapped[str | None] = mapped_column(StringID(), nullable=True)
-    bundle_version: Mapped[str] = mapped_column(StringID())
+    bundle_version: Mapped[str | None] = mapped_column(StringID(), nullable=True)
     dag_code = relationship(
         "DagCode",
         back_populates="dag_version",

airflow-core/src/airflow/models/dagbundle.py

@@ -43,7 +43,7 @@ class DagBundleModel(Base, LoggingMixin):
 
     __tablename__ = "dag_bundle"
     name: Mapped[str] = mapped_column(StringID(length=250), primary_key=True, nullable=False)
-    active: Mapped[bool] = mapped_column(Boolean, default=True)
+    active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
     version: Mapped[str | None] = mapped_column(String(200), nullable=True)
     last_refreshed: Mapped[UtcDateTime | None] = mapped_column(UtcDateTime, nullable=True)
     signed_url_template: Mapped[str | None] = mapped_column(String(200), nullable=True)

airflow-core/src/airflow/models/dagrun.py

@@ -149,29 +149,31 @@ class DagRun(Base, LoggingMixin):
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True)
     dag_id: Mapped[str] = mapped_column(StringID(), nullable=False)
-    queued_at: Mapped[UtcDateTime] = mapped_column(UtcDateTime)
+    queued_at: Mapped[UtcDateTime | None] = mapped_column(UtcDateTime, nullable=True)
     logical_date: Mapped[UtcDateTime | None] = mapped_column(UtcDateTime, nullable=True)
-    start_date: Mapped[UtcDateTime] = mapped_column(UtcDateTime)
-    end_date: Mapped[UtcDateTime] = mapped_column(UtcDateTime)
+    start_date: Mapped[UtcDateTime | None] = mapped_column(UtcDateTime, nullable=True)
+    end_date: Mapped[UtcDateTime | None] = mapped_column(UtcDateTime, nullable=True)
     _state: Mapped[str] = mapped_column("state", String(50), default=DagRunState.QUEUED)
     run_id: Mapped[str] = mapped_column(StringID(), nullable=False)
-    creating_job_id: Mapped[int] = mapped_column(Integer)
+    creating_job_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
     run_type: Mapped[str] = mapped_column(String(50), nullable=False)
-    triggered_by: Mapped[DagRunTriggeredByType] = mapped_column(
-        Enum(DagRunTriggeredByType, native_enum=False, length=50)
+    triggered_by: Mapped[DagRunTriggeredByType | None] = mapped_column(
+        Enum(DagRunTriggeredByType, native_enum=False, length=50), nullable=True
     )  # Airflow component that triggered the run.
     triggering_user_name: Mapped[str | None] = mapped_column(
         String(512),
         nullable=True,
     )  # The user that triggered the DagRun, if applicable
-    conf: Mapped[dict[str, Any]] = mapped_column(JSON().with_variant(postgresql.JSONB, "postgresql"))
+    conf: Mapped[dict[str, Any] | None] = mapped_column(
+        JSON().with_variant(postgresql.JSONB, "postgresql"), nullable=True
+    )
     # These two must be either both NULL or both datetime.
-    data_interval_start: Mapped[UtcDateTime] = mapped_column(UtcDateTime)
-    data_interval_end: Mapped[UtcDateTime] = mapped_column(UtcDateTime)
+    data_interval_start: Mapped[UtcDateTime | None] = mapped_column(UtcDateTime, nullable=True)
+    data_interval_end: Mapped[UtcDateTime | None] = mapped_column(UtcDateTime, nullable=True)
     # Earliest time when this DagRun can start running.
     run_after: Mapped[UtcDateTime] = mapped_column(UtcDateTime, default=_default_run_after, nullable=False)
     # When a scheduler last attempted to schedule TIs for this DagRun
-    last_scheduling_decision: Mapped[UtcDateTime] = mapped_column(UtcDateTime)
+    last_scheduling_decision: Mapped[UtcDateTime | None] = mapped_column(UtcDateTime, nullable=True)
     # Foreign key to LogTemplate. DagRun rows created prior to this column's
     # existence have this set to NULL. Later rows automatically populate this on
     # insert to point to the latest LogTemplate entry.
@@ -193,11 +195,13 @@ class DagRun(Base, LoggingMixin):
 
     It's possible this could change if e.g. the dag run is cleared to be rerun, or perhaps re-backfilled.
     """
-    bundle_version: Mapped[str] = mapped_column(StringID())
+    bundle_version: Mapped[str | None] = mapped_column(StringID(), nullable=True)
 
-    scheduled_by_job_id: Mapped[int] = mapped_column(Integer)
+    scheduled_by_job_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
     # Span context carrier, used for context propagation.
-    context_carrier: Mapped[dict[str, Any]] = mapped_column(MutableDict.as_mutable(ExtendedJSON))
+    context_carrier: Mapped[dict[str, Any] | None] = mapped_column(
+        MutableDict.as_mutable(ExtendedJSON), nullable=True
+    )
     span_status: Mapped[str] = mapped_column(
         String(250), server_default=SpanStatus.NOT_STARTED, nullable=False
     )

airflow-core/src/airflow/models/db_callback_request.py

@@ -21,14 +21,13 @@
 from typing import TYPE_CHECKING
 
 from sqlalchemy import Integer, String
+from sqlalchemy.orm import Mapped
 
 from airflow._shared.timezones import timezone
 from airflow.models.base import Base
 from airflow.utils.sqlalchemy import ExtendedJSON, UtcDateTime, mapped_column
 
 if TYPE_CHECKING:
-    from sqlalchemy.orm import Mapped
-
     from airflow.callbacks.callback_requests import CallbackRequest
 
 

airflow-core/src/airflow/models/deadline.py

@@ -100,7 +100,9 @@ class Deadline(Base):
     id: Mapped[str] = mapped_column(UUIDType(binary=False), primary_key=True, default=uuid6.uuid7)
 
     # If the Deadline Alert is for a DAG, store the DAG run ID from the dag_run.
-    dagrun_id: Mapped[int] = mapped_column(Integer, ForeignKey("dag_run.id", ondelete="CASCADE"))
+    dagrun_id: Mapped[int | None] = mapped_column(
+        Integer, ForeignKey("dag_run.id", ondelete="CASCADE"), nullable=True
+    )
 
     # The time after which the Deadline has passed and the callback should be triggered.
     deadline_time: Mapped[UtcDateTime] = mapped_column(UtcDateTime, nullable=False)
@@ -109,7 +111,7 @@ class Deadline(Base):
         "callback", sqlalchemy_jsonfield.JSONField(json=json), nullable=False
     )
     # The state of the deadline callback
-    callback_state: Mapped[str] = mapped_column(String(20))
+    callback_state: Mapped[str | None] = mapped_column(String(20), nullable=True)
 
     dagrun = relationship("DagRun", back_populates="deadlines")
 

airflow-core/src/airflow/models/errors.py

@@ -17,27 +17,23 @@
 # under the License.
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
-
 from sqlalchemy import Integer, String, Text
+from sqlalchemy.orm import Mapped
 
 from airflow.dag_processing.bundles.manager import DagBundlesManager
 from airflow.models.base import Base, StringID
 from airflow.utils.sqlalchemy import UtcDateTime, mapped_column
 
-if TYPE_CHECKING:
-    from sqlalchemy.orm import Mapped
-
 
 class ParseImportError(Base):
     """Stores all Import Errors which are recorded when parsing DAGs and displayed on the Webserver."""
 
     __tablename__ = "import_error"
     id: Mapped[int] = mapped_column(Integer, primary_key=True)
-    timestamp: Mapped[UtcDateTime] = mapped_column(UtcDateTime)
-    filename: Mapped[str] = mapped_column(String(1024))
-    bundle_name: Mapped[str] = mapped_column(StringID())
-    stacktrace: Mapped[str] = mapped_column(Text)
+    timestamp: Mapped[UtcDateTime | None] = mapped_column(UtcDateTime, nullable=True)
+    filename: Mapped[str | None] = mapped_column(String(1024), nullable=True)
+    bundle_name: Mapped[str | None] = mapped_column(StringID(), nullable=True)
+    stacktrace: Mapped[str | None] = mapped_column(Text, nullable=True)
 
     def full_file_path(self) -> str:
         """Return the full file path of the dag."""