We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
API Endpoint - CRUD - DAG
Hello 

We need to create several endpoints that perform basic CRUD operations on **DAG**. We need the following endpoints:

- PATCH /dags/{dag_id}

It depends on https://github.com/apache/airflow/issues/8128

Detailed information is available in the issue:
https://github.com/apache/airflow/issues/8118

LOVE,

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `airflow/api_connexion/schemas/dag_schema.py`
Content:
```
1 # Licensed to the Apache Software Foundation (ASF) under one
2 # or more contributor license agreements.  See the NOTICE file
3 # distributed with this work for additional information
4 # regarding copyright ownership.  The ASF licenses this file
5 # to you under the Apache License, Version 2.0 (the
6 # "License"); you may not use this file except in compliance
7 # with the License.  You may obtain a copy of the License at
8 #
9 #   http://www.apache.org/licenses/LICENSE-2.0
10 #
11 # Unless required by applicable law or agreed to in writing,
12 # software distributed under the License is distributed on an
13 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14 # KIND, either express or implied.  See the License for the
15 # specific language governing permissions and limitations
16 # under the License.
17 
18 from typing import List, NamedTuple
19 
20 from marshmallow import Schema, fields
21 from marshmallow_sqlalchemy import SQLAlchemySchema, auto_field
22 
23 from airflow.api_connexion.schemas.common_schema import ScheduleIntervalSchema, TimeDeltaSchema, TimezoneField
24 from airflow.models.dag import DagModel, DagTag
25 
26 
27 class DagTagSchema(SQLAlchemySchema):
28     """Dag Tag schema"""
29     class Meta:
30         """Meta"""
31 
32         model = DagTag
33 
34     name = auto_field()
35 
36 
37 class DAGSchema(SQLAlchemySchema):
38     """DAG schema"""
39 
40     class Meta:
41         """Meta"""
42         model = DagModel
43 
44     dag_id = auto_field(dump_only=True)
45     root_dag_id = auto_field(dump_only=True)
46     is_paused = auto_field(dump_only=True)
47     is_subdag = auto_field(dump_only=True)
48     fileloc = auto_field(dump_only=True)
49     owners = fields.Method("get_owners", dump_only=True)
50     description = auto_field(dump_only=True)
51     schedule_interval = fields.Nested(ScheduleIntervalSchema, dump_only=True)
52     tags = fields.List(fields.Nested(DagTagSchema), dump_only=True)
53 
54     @staticmethod
55     def get_owners(obj: DagModel):
56         """Convert owners attribute to DAG representation"""
57 
58         if not getattr(obj, 'owners', None):
59             return []
60         return obj.owners.split(",")
61 
62 
63 class DAGDetailSchema(DAGSchema):
64     """DAG details"""
65 
66     timezone = TimezoneField(dump_only=True)
67     catchup = fields.Boolean(dump_only=True)
68     orientation = fields.String(dump_only=True)
69     concurrency = fields.Integer(dump_only=True)
70     start_date = fields.DateTime(dump_only=True)
71     dag_run_timeout = fields.Nested(TimeDeltaSchema, dump_only=True, attribute="dagrun_timeout")
72     doc_md = fields.String(dump_only=True)
73     default_view = fields.String(dump_only=True)
74 
75 
76 class DAGCollection(NamedTuple):
77     """List of DAGs with metadata"""
78 
79     dags: List[DagModel]
80     total_entries: int
81 
82 
83 class DAGCollectionSchema(Schema):
84     """DAG Collection schema"""
85 
86     dags = fields.List(fields.Nested(DAGSchema))
87     total_entries = fields.Int()
88 
89 
90 dags_collection_schema = DAGCollectionSchema()
91 dag_schema = DAGSchema()
92 
93 dag_detail_schema = DAGDetailSchema()
94
```

Path: `airflow/api_connexion/schemas/common_schema.py`
Content:
```
1 # Licensed to the Apache Software Foundation (ASF) under one
2 # or more contributor license agreements.  See the NOTICE file
3 # distributed with this work for additional information
4 # regarding copyright ownership.  The ASF licenses this file
5 # to you under the Apache License, Version 2.0 (the
6 # "License"); you may not use this file except in compliance
7 # with the License.  You may obtain a copy of the License at
8 #
9 #   http://www.apache.org/licenses/LICENSE-2.0
10 #
11 # Unless required by applicable law or agreed to in writing,
12 # software distributed under the License is distributed on an
13 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14 # KIND, either express or implied.  See the License for the
15 # specific language governing permissions and limitations
16 # under the License.
17 
18 import datetime
19 import inspect
20 import typing
21 
22 import marshmallow
23 from dateutil import relativedelta
24 from marshmallow import Schema, fields, validate
25 from marshmallow_oneofschema import OneOfSchema
26 
27 from airflow.serialization.serialized_objects import SerializedBaseOperator
28 from airflow.utils.weight_rule import WeightRule
29 
30 
31 class CronExpression(typing.NamedTuple):
32     """Cron expression schema"""
33     value: str
34 
35 
36 class TimeDeltaSchema(Schema):
37     """Time delta schema"""
38 
39     objectType = fields.Constant("TimeDelta", data_key="__type")
40     days = fields.Integer()
41     seconds = fields.Integer()
42     microseconds = fields.Integer()
43 
44     @marshmallow.post_load
45     def make_time_delta(self, data, **kwargs):
46         """Create time delta based on data"""
47 
48         if "objectType" in data:
49             del data["objectType"]
50         return datetime.timedelta(**data)
51 
52 
53 class RelativeDeltaSchema(Schema):
54     """Relative delta schema"""
55 
56     objectType = fields.Constant("RelativeDelta", data_key="__type")
57     years = fields.Integer()
58     months = fields.Integer()
59     days = fields.Integer()
60     leapdays = fields.Integer()
61     hours = fields.Integer()
62     minutes = fields.Integer()
63     seconds = fields.Integer()
64     microseconds = fields.Integer()
65     year = fields.Integer()
66     month = fields.Integer()
67     day = fields.Integer()
68     hour = fields.Integer()
69     minute = fields.Integer()
70     second = fields.Integer()
71     microsecond = fields.Integer()
72 
73     @marshmallow.post_load
74     def make_relative_delta(self, data, **kwargs):
75         """Create relative delta based on data"""
76 
77         if "objectType" in data:
78             del data["objectType"]
79 
80         return relativedelta.relativedelta(**data)
81 
82 
83 class CronExpressionSchema(Schema):
84     """Cron expression schema"""
85 
86     objectType = fields.Constant("CronExpression", data_key="__type", required=True)
87     value = fields.String(required=True)
88 
89     @marshmallow.post_load
90     def make_cron_expression(self, data, **kwargs):
91         """Create cron expression based on data"""
92         return CronExpression(data["value"])
93 
94 
95 class ScheduleIntervalSchema(OneOfSchema):
96     """
97     Schedule interval.
98 
99     It supports the following types:
100 
101     * TimeDelta
102     * RelativeDelta
103     * CronExpression
104     """
105     type_field = "__type"
106     type_schemas = {
107         "TimeDelta": TimeDeltaSchema,
108         "RelativeDelta": RelativeDeltaSchema,
109         "CronExpression": CronExpressionSchema,
110     }
111 
112     def _dump(self, obj, update_fields=True, **kwargs):
113         if isinstance(obj, str):
114             obj = CronExpression(obj)
115 
116         return super()._dump(obj, update_fields=update_fields, **kwargs)
117 
118     def get_obj_type(self, obj):
119         """Select schema based on object type"""
120         if isinstance(obj, datetime.timedelta):
121             return "TimeDelta"
122         elif isinstance(obj, relativedelta.relativedelta):
123             return "RelativeDelta"
124         elif isinstance(obj, CronExpression):
125             return "CronExpression"
126         else:
127             raise Exception("Unknown object type: {}".format(obj.__class__.__name__))
128 
129 
130 class ColorField(fields.String):
131     """Schema for color property"""
132     def __init__(self, **metadata):
133         super().__init__(**metadata)
134         self.validators = (
135             [validate.Regexp("^#[a-fA-F0-9]{3,6}$")] + list(self.validators)
136         )
137 
138 
139 class WeightRuleField(fields.String):
140     """Schema for WeightRule"""
141     def __init__(self, **metadata):
142         super().__init__(**metadata)
143         self.validators = (
144             [validate.OneOf(WeightRule.all_weight_rules())] + list(self.validators)
145         )
146 
147 
148 class TimezoneField(fields.String):
149     """Schema for timezone"""
150 
151 
152 class ClassReferenceSchema(Schema):
153     """
154     Class reference schema.
155     """
156     module_path = fields.Method("_get_module", required=True)
157     class_name = fields.Method("_get_class_name", required=True)
158 
159     def _get_module(self, obj):
160         if isinstance(obj, SerializedBaseOperator):
161             return obj._task_module  # pylint: disable=protected-access
162         return inspect.getmodule(obj).__name__
163 
164     def _get_class_name(self, obj):
165         if isinstance(obj, SerializedBaseOperator):
166             return obj._task_type  # pylint: disable=protected-access
167         if isinstance(obj, type):
168             return obj.__name__
169         return type(obj).__name__
170
```

Path: `airflow/api_connexion/endpoints/dag_endpoint.py`
Content:
```
1 # Licensed to the Apache Software Foundation (ASF) under one
2 # or more contributor license agreements.  See the NOTICE file
3 # distributed with this work for additional information
4 # regarding copyright ownership.  The ASF licenses this file
5 # to you under the Apache License, Version 2.0 (the
6 # "License"); you may not use this file except in compliance
7 # with the License.  You may obtain a copy of the License at
8 #
9 #   http://www.apache.org/licenses/LICENSE-2.0
10 #
11 # Unless required by applicable law or agreed to in writing,
12 # software distributed under the License is distributed on an
13 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14 # KIND, either express or implied.  See the License for the
15 # specific language governing permissions and limitations
16 # under the License.
17 from flask import current_app
18 from sqlalchemy import func
19 
20 from airflow import DAG
21 from airflow.api_connexion import security
22 from airflow.api_connexion.exceptions import NotFound
23 from airflow.api_connexion.parameters import check_limit, format_parameters
24 from airflow.api_connexion.schemas.dag_schema import (
25     DAGCollection, dag_detail_schema, dag_schema, dags_collection_schema,
26 )
27 from airflow.models.dag import DagModel
28 from airflow.utils.session import provide_session
29 
30 
31 @security.requires_authentication
32 @provide_session
33 def get_dag(dag_id, session):
34     """
35     Get basic information about a DAG.
36     """
37     dag = session.query(DagModel).filter(DagModel.dag_id == dag_id).one_or_none()
38 
39     if dag is None:
40         raise NotFound("DAG not found")
41 
42     return dag_schema.dump(dag)
43 
44 
45 @security.requires_authentication
46 def get_dag_details(dag_id):
47     """
48     Get details of DAG.
49     """
50     dag: DAG = current_app.dag_bag.get_dag(dag_id)
51     if not dag:
52         raise NotFound("DAG not found")
53     return dag_detail_schema.dump(dag)
54 
55 
56 @security.requires_authentication
57 @format_parameters({
58     'limit': check_limit
59 })
60 @provide_session
61 def get_dags(session, limit, offset=0):
62     """
63     Get all DAGs.
64     """
65     dags = session.query(DagModel).order_by(DagModel.dag_id).offset(offset).limit(limit).all()
66 
67     total_entries = session.query(func.count(DagModel.dag_id)).scalar()
68 
69     return dags_collection_schema.dump(DAGCollection(dags=dags, total_entries=total_entries))
70 
71 
72 @security.requires_authentication
73 def patch_dag():
74     """
75     Update the specific DAG
76     """
77     raise NotImplementedError("Not implemented yet.")
78
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/airflow/api_connexion/endpoints/dag_endpoint.py b/airflow/api_connexion/endpoints/dag_endpoint.py
--- a/airflow/api_connexion/endpoints/dag_endpoint.py
+++ b/airflow/api_connexion/endpoints/dag_endpoint.py
@@ -14,12 +14,13 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-from flask import current_app
+from flask import current_app, request
+from marshmallow import ValidationError
 from sqlalchemy import func
 
 from airflow import DAG
 from airflow.api_connexion import security
-from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.exceptions import BadRequest, NotFound
 from airflow.api_connexion.parameters import check_limit, format_parameters
 from airflow.api_connexion.schemas.dag_schema import (
     DAGCollection, dag_detail_schema, dag_schema, dags_collection_schema,
@@ -70,8 +71,19 @@
 
 
 @security.requires_authentication
-def patch_dag():
+@provide_session
+def patch_dag(session, dag_id):
     """
     Update the specific DAG
     """
-    raise NotImplementedError("Not implemented yet.")
+    dag = session.query(DagModel).filter(DagModel.dag_id == dag_id).one_or_none()
+    if not dag:
+        raise NotFound(f"Dag with id: '{dag_id}' not found")
+    try:
+        patch_body = dag_schema.load(request.json, session=session)
+    except ValidationError as err:
+        raise BadRequest("Invalid Dag schema", detail=str(err.messages))
+    for key, value in patch_body.items():
+        setattr(dag, key, value)
+    session.commit()
+    return dag_schema.dump(dag)
diff --git a/airflow/api_connexion/schemas/common_schema.py b/airflow/api_connexion/schemas/common_schema.py
--- a/airflow/api_connexion/schemas/common_schema.py
+++ b/airflow/api_connexion/schemas/common_schema.py
@@ -83,7 +83,7 @@
 class CronExpressionSchema(Schema):
     """Cron expression schema"""
 
-    objectType = fields.Constant("CronExpression", data_key="__type", required=True)
+    objectType = fields.Constant("CronExpression", data_key="__type")
     value = fields.String(required=True)
 
     @marshmallow.post_load
diff --git a/airflow/api_connexion/schemas/dag_schema.py b/airflow/api_connexion/schemas/dag_schema.py
--- a/airflow/api_connexion/schemas/dag_schema.py
+++ b/airflow/api_connexion/schemas/dag_schema.py
@@ -43,12 +43,12 @@
 
     dag_id = auto_field(dump_only=True)
     root_dag_id = auto_field(dump_only=True)
-    is_paused = auto_field(dump_only=True)
+    is_paused = auto_field()
     is_subdag = auto_field(dump_only=True)
     fileloc = auto_field(dump_only=True)
     owners = fields.Method("get_owners", dump_only=True)
     description = auto_field(dump_only=True)
-    schedule_interval = fields.Nested(ScheduleIntervalSchema, dump_only=True)
+    schedule_interval = fields.Nested(ScheduleIntervalSchema)
     tags = fields.List(fields.Nested(DagTagSchema), dump_only=True)
 
     @staticmethod

{"golden_diff": "diff --git a/airflow/api_connexion/endpoints/dag_endpoint.py b/airflow/api_connexion/endpoints/dag_endpoint.py\n--- a/airflow/api_connexion/endpoints/dag_endpoint.py\n+++ b/airflow/api_connexion/endpoints/dag_endpoint.py\n@@ -14,12 +14,13 @@\n # KIND, either express or implied.  See the License for the\n # specific language governing permissions and limitations\n # under the License.\n-from flask import current_app\n+from flask import current_app, request\n+from marshmallow import ValidationError\n from sqlalchemy import func\n \n from airflow import DAG\n from airflow.api_connexion import security\n-from airflow.api_connexion.exceptions import NotFound\n+from airflow.api_connexion.exceptions import BadRequest, NotFound\n from airflow.api_connexion.parameters import check_limit, format_parameters\n from airflow.api_connexion.schemas.dag_schema import (\n     DAGCollection, dag_detail_schema, dag_schema, dags_collection_schema,\n@@ -70,8 +71,19 @@\n \n \n @security.requires_authentication\n-def patch_dag():\n+@provide_session\n+def patch_dag(session, dag_id):\n     \"\"\"\n     Update the specific DAG\n     \"\"\"\n-    raise NotImplementedError(\"Not implemented yet.\")\n+    dag = session.query(DagModel).filter(DagModel.dag_id == dag_id).one_or_none()\n+    if not dag:\n+        raise NotFound(f\"Dag with id: '{dag_id}' not found\")\n+    try:\n+        patch_body = dag_schema.load(request.json, session=session)\n+    except ValidationError as err:\n+        raise BadRequest(\"Invalid Dag schema\", detail=str(err.messages))\n+    for key, value in patch_body.items():\n+        setattr(dag, key, value)\n+    session.commit()\n+    return dag_schema.dump(dag)\ndiff --git a/airflow/api_connexion/schemas/common_schema.py b/airflow/api_connexion/schemas/common_schema.py\n--- a/airflow/api_connexion/schemas/common_schema.py\n+++ b/airflow/api_connexion/schemas/common_schema.py\n@@ -83,7 +83,7 @@\n class CronExpressionSchema(Schema):\n     \"\"\"Cron expression schema\"\"\"\n \n-    objectType = fields.Constant(\"CronExpression\", data_key=\"__type\", required=True)\n+    objectType = fields.Constant(\"CronExpression\", data_key=\"__type\")\n     value = fields.String(required=True)\n \n     @marshmallow.post_load\ndiff --git a/airflow/api_connexion/schemas/dag_schema.py b/airflow/api_connexion/schemas/dag_schema.py\n--- a/airflow/api_connexion/schemas/dag_schema.py\n+++ b/airflow/api_connexion/schemas/dag_schema.py\n@@ -43,12 +43,12 @@\n \n     dag_id = auto_field(dump_only=True)\n     root_dag_id = auto_field(dump_only=True)\n-    is_paused = auto_field(dump_only=True)\n+    is_paused = auto_field()\n     is_subdag = auto_field(dump_only=True)\n     fileloc = auto_field(dump_only=True)\n     owners = fields.Method(\"get_owners\", dump_only=True)\n     description = auto_field(dump_only=True)\n-    schedule_interval = fields.Nested(ScheduleIntervalSchema, dump_only=True)\n+    schedule_interval = fields.Nested(ScheduleIntervalSchema)\n     tags = fields.List(fields.Nested(DagTagSchema), dump_only=True)\n \n     @staticmethod\n", "issue": "API Endpoint - CRUD - DAG\nHello \r\n\r\nWe need to create several endpoints that perform basic CRUD operations on **DAG**. We need the following endpoints:\r\n\r\n- PATCH /dags/{dag_id}\r\n\r\nIt depends on https://github.com/apache/airflow/issues/8128\r\n\r\nDetailed information is available in the issue:\r\nhttps://github.com/apache/airflow/issues/8118\r\n\r\nLOVE,\n", "before_files": [{"content": "# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\nfrom typing import List, NamedTuple\n\nfrom marshmallow import Schema, fields\nfrom marshmallow_sqlalchemy import SQLAlchemySchema, auto_field\n\nfrom airflow.api_connexion.schemas.common_schema import ScheduleIntervalSchema, TimeDeltaSchema, TimezoneField\nfrom airflow.models.dag import DagModel, DagTag\n\n\nclass DagTagSchema(SQLAlchemySchema):\n    \"\"\"Dag Tag schema\"\"\"\n    class Meta:\n        \"\"\"Meta\"\"\"\n\n        model = DagTag\n\n    name = auto_field()\n\n\nclass DAGSchema(SQLAlchemySchema):\n    \"\"\"DAG schema\"\"\"\n\n    class Meta:\n        \"\"\"Meta\"\"\"\n        model = DagModel\n\n    dag_id = auto_field(dump_only=True)\n    root_dag_id = auto_field(dump_only=True)\n    is_paused = auto_field(dump_only=True)\n    is_subdag = auto_field(dump_only=True)\n    fileloc = auto_field(dump_only=True)\n    owners = fields.Method(\"get_owners\", dump_only=True)\n    description = auto_field(dump_only=True)\n    schedule_interval = fields.Nested(ScheduleIntervalSchema, dump_only=True)\n    tags = fields.List(fields.Nested(DagTagSchema), dump_only=True)\n\n    @staticmethod\n    def get_owners(obj: DagModel):\n        \"\"\"Convert owners attribute to DAG representation\"\"\"\n\n        if not getattr(obj, 'owners', None):\n            return []\n        return obj.owners.split(\",\")\n\n\nclass DAGDetailSchema(DAGSchema):\n    \"\"\"DAG details\"\"\"\n\n    timezone = TimezoneField(dump_only=True)\n    catchup = fields.Boolean(dump_only=True)\n    orientation = fields.String(dump_only=True)\n    concurrency = fields.Integer(dump_only=True)\n    start_date = fields.DateTime(dump_only=True)\n    dag_run_timeout = fields.Nested(TimeDeltaSchema, dump_only=True, attribute=\"dagrun_timeout\")\n    doc_md = fields.String(dump_only=True)\n    default_view = fields.String(dump_only=True)\n\n\nclass DAGCollection(NamedTuple):\n    \"\"\"List of DAGs with metadata\"\"\"\n\n    dags: List[DagModel]\n    total_entries: int\n\n\nclass DAGCollectionSchema(Schema):\n    \"\"\"DAG Collection schema\"\"\"\n\n    dags = fields.List(fields.Nested(DAGSchema))\n    total_entries = fields.Int()\n\n\ndags_collection_schema = DAGCollectionSchema()\ndag_schema = DAGSchema()\n\ndag_detail_schema = DAGDetailSchema()\n", "path": "airflow/api_connexion/schemas/dag_schema.py"}, {"content": "# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\nimport datetime\nimport inspect\nimport typing\n\nimport marshmallow\nfrom dateutil import relativedelta\nfrom marshmallow import Schema, fields, validate\nfrom marshmallow_oneofschema import OneOfSchema\n\nfrom airflow.serialization.serialized_objects import SerializedBaseOperator\nfrom airflow.utils.weight_rule import WeightRule\n\n\nclass CronExpression(typing.NamedTuple):\n    \"\"\"Cron expression schema\"\"\"\n    value: str\n\n\nclass TimeDeltaSchema(Schema):\n    \"\"\"Time delta schema\"\"\"\n\n    objectType = fields.Constant(\"TimeDelta\", data_key=\"__type\")\n    days = fields.Integer()\n    seconds = fields.Integer()\n    microseconds = fields.Integer()\n\n    @marshmallow.post_load\n    def make_time_delta(self, data, **kwargs):\n        \"\"\"Create time delta based on data\"\"\"\n\n        if \"objectType\" in data:\n            del data[\"objectType\"]\n        return datetime.timedelta(**data)\n\n\nclass RelativeDeltaSchema(Schema):\n    \"\"\"Relative delta schema\"\"\"\n\n    objectType = fields.Constant(\"RelativeDelta\", data_key=\"__type\")\n    years = fields.Integer()\n    months = fields.Integer()\n    days = fields.Integer()\n    leapdays = fields.Integer()\n    hours = fields.Integer()\n    minutes = fields.Integer()\n    seconds = fields.Integer()\n    microseconds = fields.Integer()\n    year = fields.Integer()\n    month = fields.Integer()\n    day = fields.Integer()\n    hour = fields.Integer()\n    minute = fields.Integer()\n    second = fields.Integer()\n    microsecond = fields.Integer()\n\n    @marshmallow.post_load\n    def make_relative_delta(self, data, **kwargs):\n        \"\"\"Create relative delta based on data\"\"\"\n\n        if \"objectType\" in data:\n            del data[\"objectType\"]\n\n        return relativedelta.relativedelta(**data)\n\n\nclass CronExpressionSchema(Schema):\n    \"\"\"Cron expression schema\"\"\"\n\n    objectType = fields.Constant(\"CronExpression\", data_key=\"__type\", required=True)\n    value = fields.String(required=True)\n\n    @marshmallow.post_load\n    def make_cron_expression(self, data, **kwargs):\n        \"\"\"Create cron expression based on data\"\"\"\n        return CronExpression(data[\"value\"])\n\n\nclass ScheduleIntervalSchema(OneOfSchema):\n    \"\"\"\n    Schedule interval.\n\n    It supports the following types:\n\n    * TimeDelta\n    * RelativeDelta\n    * CronExpression\n    \"\"\"\n    type_field = \"__type\"\n    type_schemas = {\n        \"TimeDelta\": TimeDeltaSchema,\n        \"RelativeDelta\": RelativeDeltaSchema,\n        \"CronExpression\": CronExpressionSchema,\n    }\n\n    def _dump(self, obj, update_fields=True, **kwargs):\n        if isinstance(obj, str):\n            obj = CronExpression(obj)\n\n        return super()._dump(obj, update_fields=update_fields, **kwargs)\n\n    def get_obj_type(self, obj):\n        \"\"\"Select schema based on object type\"\"\"\n        if isinstance(obj, datetime.timedelta):\n            return \"TimeDelta\"\n        elif isinstance(obj, relativedelta.relativedelta):\n            return \"RelativeDelta\"\n        elif isinstance(obj, CronExpression):\n            return \"CronExpression\"\n        else:\n            raise Exception(\"Unknown object type: {}\".format(obj.__class__.__name__))\n\n\nclass ColorField(fields.String):\n    \"\"\"Schema for color property\"\"\"\n    def __init__(self, **metadata):\n        super().__init__(**metadata)\n        self.validators = (\n            [validate.Regexp(\"^#[a-fA-F0-9]{3,6}$\")] + list(self.validators)\n        )\n\n\nclass WeightRuleField(fields.String):\n    \"\"\"Schema for WeightRule\"\"\"\n    def __init__(self, **metadata):\n        super().__init__(**metadata)\n        self.validators = (\n            [validate.OneOf(WeightRule.all_weight_rules())] + list(self.validators)\n        )\n\n\nclass TimezoneField(fields.String):\n    \"\"\"Schema for timezone\"\"\"\n\n\nclass ClassReferenceSchema(Schema):\n    \"\"\"\n    Class reference schema.\n    \"\"\"\n    module_path = fields.Method(\"_get_module\", required=True)\n    class_name = fields.Method(\"_get_class_name\", required=True)\n\n    def _get_module(self, obj):\n        if isinstance(obj, SerializedBaseOperator):\n            return obj._task_module  # pylint: disable=protected-access\n        return inspect.getmodule(obj).__name__\n\n    def _get_class_name(self, obj):\n        if isinstance(obj, SerializedBaseOperator):\n            return obj._task_type  # pylint: disable=protected-access\n        if isinstance(obj, type):\n            return obj.__name__\n        return type(obj).__name__\n", "path": "airflow/api_connexion/schemas/common_schema.py"}, {"content": "# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\nfrom flask import current_app\nfrom sqlalchemy import func\n\nfrom airflow import DAG\nfrom airflow.api_connexion import security\nfrom airflow.api_connexion.exceptions import NotFound\nfrom airflow.api_connexion.parameters import check_limit, format_parameters\nfrom airflow.api_connexion.schemas.dag_schema import (\n    DAGCollection, dag_detail_schema, dag_schema, dags_collection_schema,\n)\nfrom airflow.models.dag import DagModel\nfrom airflow.utils.session import provide_session\n\n\n@security.requires_authentication\n@provide_session\ndef get_dag(dag_id, session):\n    \"\"\"\n    Get basic information about a DAG.\n    \"\"\"\n    dag = session.query(DagModel).filter(DagModel.dag_id == dag_id).one_or_none()\n\n    if dag is None:\n        raise NotFound(\"DAG not found\")\n\n    return dag_schema.dump(dag)\n\n\n@security.requires_authentication\ndef get_dag_details(dag_id):\n    \"\"\"\n    Get details of DAG.\n    \"\"\"\n    dag: DAG = current_app.dag_bag.get_dag(dag_id)\n    if not dag:\n        raise NotFound(\"DAG not found\")\n    return dag_detail_schema.dump(dag)\n\n\n@security.requires_authentication\n@format_parameters({\n    'limit': check_limit\n})\n@provide_session\ndef get_dags(session, limit, offset=0):\n    \"\"\"\n    Get all DAGs.\n    \"\"\"\n    dags = session.query(DagModel).order_by(DagModel.dag_id).offset(offset).limit(limit).all()\n\n    total_entries = session.query(func.count(DagModel.dag_id)).scalar()\n\n    return dags_collection_schema.dump(DAGCollection(dags=dags, total_entries=total_entries))\n\n\n@security.requires_authentication\ndef patch_dag():\n    \"\"\"\n    Update the specific DAG\n    \"\"\"\n    raise NotImplementedError(\"Not implemented yet.\")\n", "path": "airflow/api_connexion/endpoints/dag_endpoint.py"}], "after_files": [{"content": "# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\nfrom typing import List, NamedTuple\n\nfrom marshmallow import Schema, fields\nfrom marshmallow_sqlalchemy import SQLAlchemySchema, auto_field\n\nfrom airflow.api_connexion.schemas.common_schema import ScheduleIntervalSchema, TimeDeltaSchema, TimezoneField\nfrom airflow.models.dag import DagModel, DagTag\n\n\nclass DagTagSchema(SQLAlchemySchema):\n    \"\"\"Dag Tag schema\"\"\"\n    class Meta:\n        \"\"\"Meta\"\"\"\n\n        model = DagTag\n\n    name = auto_field()\n\n\nclass DAGSchema(SQLAlchemySchema):\n    \"\"\"DAG schema\"\"\"\n\n    class Meta:\n        \"\"\"Meta\"\"\"\n        model = DagModel\n\n    dag_id = auto_field(dump_only=True)\n    root_dag_id = auto_field(dump_only=True)\n    is_paused = auto_field()\n    is_subdag = auto_field(dump_only=True)\n    fileloc = auto_field(dump_only=True)\n    owners = fields.Method(\"get_owners\", dump_only=True)\n    description = auto_field(dump_only=True)\n    schedule_interval = fields.Nested(ScheduleIntervalSchema)\n    tags = fields.List(fields.Nested(DagTagSchema), dump_only=True)\n\n    @staticmethod\n    def get_owners(obj: DagModel):\n        \"\"\"Convert owners attribute to DAG representation\"\"\"\n\n        if not getattr(obj, 'owners', None):\n            return []\n        return obj.owners.split(\",\")\n\n\nclass DAGDetailSchema(DAGSchema):\n    \"\"\"DAG details\"\"\"\n\n    timezone = TimezoneField(dump_only=True)\n    catchup = fields.Boolean(dump_only=True)\n    orientation = fields.String(dump_only=True)\n    concurrency = fields.Integer(dump_only=True)\n    start_date = fields.DateTime(dump_only=True)\n    dag_run_timeout = fields.Nested(TimeDeltaSchema, dump_only=True, attribute=\"dagrun_timeout\")\n    doc_md = fields.String(dump_only=True)\n    default_view = fields.String(dump_only=True)\n\n\nclass DAGCollection(NamedTuple):\n    \"\"\"List of DAGs with metadata\"\"\"\n\n    dags: List[DagModel]\n    total_entries: int\n\n\nclass DAGCollectionSchema(Schema):\n    \"\"\"DAG Collection schema\"\"\"\n\n    dags = fields.List(fields.Nested(DAGSchema))\n    total_entries = fields.Int()\n\n\ndags_collection_schema = DAGCollectionSchema()\ndag_schema = DAGSchema()\n\ndag_detail_schema = DAGDetailSchema()\n", "path": "airflow/api_connexion/schemas/dag_schema.py"}, {"content": "# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\nimport datetime\nimport inspect\nimport typing\n\nimport marshmallow\nfrom dateutil import relativedelta\nfrom marshmallow import Schema, fields, validate\nfrom marshmallow_oneofschema import OneOfSchema\n\nfrom airflow.serialization.serialized_objects import SerializedBaseOperator\nfrom airflow.utils.weight_rule import WeightRule\n\n\nclass CronExpression(typing.NamedTuple):\n    \"\"\"Cron expression schema\"\"\"\n    value: str\n\n\nclass TimeDeltaSchema(Schema):\n    \"\"\"Time delta schema\"\"\"\n\n    objectType = fields.Constant(\"TimeDelta\", data_key=\"__type\")\n    days = fields.Integer()\n    seconds = fields.Integer()\n    microseconds = fields.Integer()\n\n    @marshmallow.post_load\n    def make_time_delta(self, data, **kwargs):\n        \"\"\"Create time delta based on data\"\"\"\n\n        if \"objectType\" in data:\n            del data[\"objectType\"]\n        return datetime.timedelta(**data)\n\n\nclass RelativeDeltaSchema(Schema):\n    \"\"\"Relative delta schema\"\"\"\n\n    objectType = fields.Constant(\"RelativeDelta\", data_key=\"__type\")\n    years = fields.Integer()\n    months = fields.Integer()\n    days = fields.Integer()\n    leapdays = fields.Integer()\n    hours = fields.Integer()\n    minutes = fields.Integer()\n    seconds = fields.Integer()\n    microseconds = fields.Integer()\n    year = fields.Integer()\n    month = fields.Integer()\n    day = fields.Integer()\n    hour = fields.Integer()\n    minute = fields.Integer()\n    second = fields.Integer()\n    microsecond = fields.Integer()\n\n    @marshmallow.post_load\n    def make_relative_delta(self, data, **kwargs):\n        \"\"\"Create relative delta based on data\"\"\"\n\n        if \"objectType\" in data:\n            del data[\"objectType\"]\n\n        return relativedelta.relativedelta(**data)\n\n\nclass CronExpressionSchema(Schema):\n    \"\"\"Cron expression schema\"\"\"\n\n    objectType = fields.Constant(\"CronExpression\", data_key=\"__type\")\n    value = fields.String(required=True)\n\n    @marshmallow.post_load\n    def make_cron_expression(self, data, **kwargs):\n        \"\"\"Create cron expression based on data\"\"\"\n        return CronExpression(data[\"value\"])\n\n\nclass ScheduleIntervalSchema(OneOfSchema):\n    \"\"\"\n    Schedule interval.\n\n    It supports the following types:\n\n    * TimeDelta\n    * RelativeDelta\n    * CronExpression\n    \"\"\"\n    type_field = \"__type\"\n    type_schemas = {\n        \"TimeDelta\": TimeDeltaSchema,\n        \"RelativeDelta\": RelativeDeltaSchema,\n        \"CronExpression\": CronExpressionSchema,\n    }\n\n    def _dump(self, obj, update_fields=True, **kwargs):\n        if isinstance(obj, str):\n            obj = CronExpression(obj)\n\n        return super()._dump(obj, update_fields=update_fields, **kwargs)\n\n    def get_obj_type(self, obj):\n        \"\"\"Select schema based on object type\"\"\"\n        if isinstance(obj, datetime.timedelta):\n            return \"TimeDelta\"\n        elif isinstance(obj, relativedelta.relativedelta):\n            return \"RelativeDelta\"\n        elif isinstance(obj, CronExpression):\n            return \"CronExpression\"\n        else:\n            raise Exception(\"Unknown object type: {}\".format(obj.__class__.__name__))\n\n\nclass ColorField(fields.String):\n    \"\"\"Schema for color property\"\"\"\n    def __init__(self, **metadata):\n        super().__init__(**metadata)\n        self.validators = (\n            [validate.Regexp(\"^#[a-fA-F0-9]{3,6}$\")] + list(self.validators)\n        )\n\n\nclass WeightRuleField(fields.String):\n    \"\"\"Schema for WeightRule\"\"\"\n    def __init__(self, **metadata):\n        super().__init__(**metadata)\n        self.validators = (\n            [validate.OneOf(WeightRule.all_weight_rules())] + list(self.validators)\n        )\n\n\nclass TimezoneField(fields.String):\n    \"\"\"Schema for timezone\"\"\"\n\n\nclass ClassReferenceSchema(Schema):\n    \"\"\"\n    Class reference schema.\n    \"\"\"\n    module_path = fields.Method(\"_get_module\", required=True)\n    class_name = fields.Method(\"_get_class_name\", required=True)\n\n    def _get_module(self, obj):\n        if isinstance(obj, SerializedBaseOperator):\n            return obj._task_module  # pylint: disable=protected-access\n        return inspect.getmodule(obj).__name__\n\n    def _get_class_name(self, obj):\n        if isinstance(obj, SerializedBaseOperator):\n            return obj._task_type  # pylint: disable=protected-access\n        if isinstance(obj, type):\n            return obj.__name__\n        return type(obj).__name__\n", "path": "airflow/api_connexion/schemas/common_schema.py"}, {"content": "# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\nfrom flask import current_app, request\nfrom marshmallow import ValidationError\nfrom sqlalchemy import func\n\nfrom airflow import DAG\nfrom airflow.api_connexion import security\nfrom airflow.api_connexion.exceptions import BadRequest, NotFound\nfrom airflow.api_connexion.parameters import check_limit, format_parameters\nfrom airflow.api_connexion.schemas.dag_schema import (\n    DAGCollection, dag_detail_schema, dag_schema, dags_collection_schema,\n)\nfrom airflow.models.dag import DagModel\nfrom airflow.utils.session import provide_session\n\n\n@security.requires_authentication\n@provide_session\ndef get_dag(dag_id, session):\n    \"\"\"\n    Get basic information about a DAG.\n    \"\"\"\n    dag = session.query(DagModel).filter(DagModel.dag_id == dag_id).one_or_none()\n\n    if dag is None:\n        raise NotFound(\"DAG not found\")\n\n    return dag_schema.dump(dag)\n\n\n@security.requires_authentication\ndef get_dag_details(dag_id):\n    \"\"\"\n    Get details of DAG.\n    \"\"\"\n    dag: DAG = current_app.dag_bag.get_dag(dag_id)\n    if not dag:\n        raise NotFound(\"DAG not found\")\n    return dag_detail_schema.dump(dag)\n\n\n@security.requires_authentication\n@format_parameters({\n    'limit': check_limit\n})\n@provide_session\ndef get_dags(session, limit, offset=0):\n    \"\"\"\n    Get all DAGs.\n    \"\"\"\n    dags = session.query(DagModel).order_by(DagModel.dag_id).offset(offset).limit(limit).all()\n\n    total_entries = session.query(func.count(DagModel.dag_id)).scalar()\n\n    return dags_collection_schema.dump(DAGCollection(dags=dags, total_entries=total_entries))\n\n\n@security.requires_authentication\n@provide_session\ndef patch_dag(session, dag_id):\n    \"\"\"\n    Update the specific DAG\n    \"\"\"\n    dag = session.query(DagModel).filter(DagModel.dag_id == dag_id).one_or_none()\n    if not dag:\n        raise NotFound(f\"Dag with id: '{dag_id}' not found\")\n    try:\n        patch_body = dag_schema.load(request.json, session=session)\n    except ValidationError as err:\n        raise BadRequest(\"Invalid Dag schema\", detail=str(err.messages))\n    for key, value in patch_body.items():\n        setattr(dag, key, value)\n    session.commit()\n    return dag_schema.dump(dag)\n", "path": "airflow/api_connexion/endpoints/dag_endpoint.py"}]}