We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Lets use Text type for value column of TaskParameter table
## Description

Change `value` column type from `string(256)` to `text` because of the issues on insert to task parameters table.

[error](https://gist.github.com/oivoodoo/0fb78f5a041bb08939f30f4af10c00c6) 
## Motivation and Context

Found this issue from the log file in our production system.
## Have you tested this? If so, how?

Going to test on 17.10.2016 in the next our deploy.

luigi parameter value column is set to limit size of 256
<!---
We use GitHub issues mainly for tracking bugs and feature requests.
Questions for how to use luigi can be sent to the mailing list.

Currently, there are no strict procedures or guidelines for submitting issues.
In short, please just use common sense.

Common sense includes this at bare-minimum:

 * search for similar issues posted before creating a new issue.
 * Use markdown to format all code/logs. Issues which are hard to read
   when rendered on GitHub might be closed with a friendly reminder of this.
 * If applicable, reading relevant parts of the documentation.

Also, add steps to reproduce the bug, if applicable. Sample code would be nice too :)

For more information on how to submit valuable contributions,
see https://opensource.guide/how-to-contribute/#how-to-submit-a-contribution
-->

But we need values longer than that. Can the limit be raised? In postgres this can just be `TEXT` type.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `luigi/db_task_history.py`
Content:
```
1 # -*- coding: utf-8 -*-
2 #
3 # Copyright 2012-2015 Spotify AB
4 #
5 # Licensed under the Apache License, Version 2.0 (the "License");
6 # you may not use this file except in compliance with the License.
7 # 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, software
12 # distributed under the License is distributed on an "AS IS" BASIS,
13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
16 #
17 
18 """
19 Provides a database backend to the central scheduler. This lets you see historical runs.
20 See :ref:`TaskHistory` for information about how to turn out the task history feature.
21 """
22 #
23 # Description: Added codes for visualization of how long each task takes
24 # running-time until it reaches the next status (failed or done)
25 # At "{base_url}/tasklist", all completed(failed or done) tasks are shown.
26 # At "{base_url}/tasklist", a user can select one specific task to see
27 # how its running-time has changed over time.
28 # At "{base_url}/tasklist/{task_name}", it visualizes a multi-bar graph
29 # that represents the changes of the running-time for a selected task
30 # up to the next status (failed or done).
31 # This visualization let us know how the running-time of the specific task
32 # has changed over time.
33 #
34 # Copyright 2015 Naver Corp.
35 # Author Yeseul Park (yeseul.park@navercorp.com)
36 #
37 
38 import datetime
39 import logging
40 from contextlib import contextmanager
41 
42 from luigi import six
43 
44 from luigi import configuration
45 from luigi import task_history
46 from luigi.task_status import DONE, FAILED, PENDING, RUNNING
47 
48 import sqlalchemy
49 import sqlalchemy.ext.declarative
50 import sqlalchemy.orm
51 import sqlalchemy.orm.collections
52 from sqlalchemy.engine import reflection
53 Base = sqlalchemy.ext.declarative.declarative_base()
54 
55 logger = logging.getLogger('luigi-interface')
56 
57 
58 class DbTaskHistory(task_history.TaskHistory):
59     """
60     Task History that writes to a database using sqlalchemy.
61     Also has methods for useful db queries.
62     """
63     CURRENT_SOURCE_VERSION = 1
64 
65     @contextmanager
66     def _session(self, session=None):
67         if session:
68             yield session
69         else:
70             session = self.session_factory()
71             try:
72                 yield session
73             except BaseException:
74                 session.rollback()
75                 raise
76             else:
77                 session.commit()
78 
79     def __init__(self):
80         config = configuration.get_config()
81         connection_string = config.get('task_history', 'db_connection')
82         self.engine = sqlalchemy.create_engine(connection_string)
83         self.session_factory = sqlalchemy.orm.sessionmaker(bind=self.engine, expire_on_commit=False)
84         Base.metadata.create_all(self.engine)
85         self.tasks = {}  # task_id -> TaskRecord
86 
87         _upgrade_schema(self.engine)
88 
89     def task_scheduled(self, task):
90         htask = self._get_task(task, status=PENDING)
91         self._add_task_event(htask, TaskEvent(event_name=PENDING, ts=datetime.datetime.now()))
92 
93     def task_finished(self, task, successful):
94         event_name = DONE if successful else FAILED
95         htask = self._get_task(task, status=event_name)
96         self._add_task_event(htask, TaskEvent(event_name=event_name, ts=datetime.datetime.now()))
97 
98     def task_started(self, task, worker_host):
99         htask = self._get_task(task, status=RUNNING, host=worker_host)
100         self._add_task_event(htask, TaskEvent(event_name=RUNNING, ts=datetime.datetime.now()))
101 
102     def _get_task(self, task, status, host=None):
103         if task.id in self.tasks:
104             htask = self.tasks[task.id]
105             htask.status = status
106             if host:
107                 htask.host = host
108         else:
109             htask = self.tasks[task.id] = task_history.StoredTask(task, status, host)
110         return htask
111 
112     def _add_task_event(self, task, event):
113         for (task_record, session) in self._find_or_create_task(task):
114             task_record.events.append(event)
115 
116     def _find_or_create_task(self, task):
117         with self._session() as session:
118             if task.record_id is not None:
119                 logger.debug("Finding task with record_id [%d]", task.record_id)
120                 task_record = session.query(TaskRecord).get(task.record_id)
121                 if not task_record:
122                     raise Exception("Task with record_id, but no matching Task record!")
123                 yield (task_record, session)
124             else:
125                 task_record = TaskRecord(task_id=task._task.id, name=task.task_family, host=task.host)
126                 for (k, v) in six.iteritems(task.parameters):
127                     task_record.parameters[k] = TaskParameter(name=k, value=v)
128                 session.add(task_record)
129                 yield (task_record, session)
130             if task.host:
131                 task_record.host = task.host
132         task.record_id = task_record.id
133 
134     def find_all_by_parameters(self, task_name, session=None, **task_params):
135         """
136         Find tasks with the given task_name and the same parameters as the kwargs.
137         """
138         with self._session(session) as session:
139             query = session.query(TaskRecord).join(TaskEvent).filter(TaskRecord.name == task_name)
140             for (k, v) in six.iteritems(task_params):
141                 alias = sqlalchemy.orm.aliased(TaskParameter)
142                 query = query.join(alias).filter(alias.name == k, alias.value == v)
143 
144             tasks = query.order_by(TaskEvent.ts)
145             for task in tasks:
146                 # Sanity check
147                 assert all(k in task.parameters and v == str(task.parameters[k].value) for (k, v) in six.iteritems(task_params))
148 
149                 yield task
150 
151     def find_all_by_name(self, task_name, session=None):
152         """
153         Find all tasks with the given task_name.
154         """
155         return self.find_all_by_parameters(task_name, session)
156 
157     def find_latest_runs(self, session=None):
158         """
159         Return tasks that have been updated in the past 24 hours.
160         """
161         with self._session(session) as session:
162             yesterday = datetime.datetime.now() - datetime.timedelta(days=1)
163             return session.query(TaskRecord).\
164                 join(TaskEvent).\
165                 filter(TaskEvent.ts >= yesterday).\
166                 group_by(TaskRecord.id, TaskEvent.event_name, TaskEvent.ts).\
167                 order_by(TaskEvent.ts.desc()).\
168                 all()
169 
170     def find_all_runs(self, session=None):
171         """
172         Return all tasks that have been updated.
173         """
174         with self._session(session) as session:
175             return session.query(TaskRecord).all()
176 
177     def find_all_events(self, session=None):
178         """
179         Return all running/failed/done events.
180         """
181         with self._session(session) as session:
182             return session.query(TaskEvent).all()
183 
184     def find_task_by_id(self, id, session=None):
185         """
186         Find task with the given record ID.
187         """
188         with self._session(session) as session:
189             return session.query(TaskRecord).get(id)
190 
191 
192 class TaskParameter(Base):
193     """
194     Table to track luigi.Parameter()s of a Task.
195     """
196     __tablename__ = 'task_parameters'
197     task_id = sqlalchemy.Column(sqlalchemy.Integer, sqlalchemy.ForeignKey('tasks.id'), primary_key=True)
198     name = sqlalchemy.Column(sqlalchemy.String(128), primary_key=True)
199     value = sqlalchemy.Column(sqlalchemy.String(256))
200 
201     def __repr__(self):
202         return "TaskParameter(task_id=%d, name=%s, value=%s)" % (self.task_id, self.name, self.value)
203 
204 
205 class TaskEvent(Base):
206     """
207     Table to track when a task is scheduled, starts, finishes, and fails.
208     """
209     __tablename__ = 'task_events'
210     id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True)
211     task_id = sqlalchemy.Column(sqlalchemy.Integer, sqlalchemy.ForeignKey('tasks.id'), index=True)
212     event_name = sqlalchemy.Column(sqlalchemy.String(20))
213     ts = sqlalchemy.Column(sqlalchemy.TIMESTAMP, index=True, nullable=False)
214 
215     def __repr__(self):
216         return "TaskEvent(task_id=%s, event_name=%s, ts=%s" % (self.task_id, self.event_name, self.ts)
217 
218 
219 class TaskRecord(Base):
220     """
221     Base table to track information about a luigi.Task.
222 
223     References to other tables are available through task.events, task.parameters, etc.
224     """
225     __tablename__ = 'tasks'
226     id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True)
227     task_id = sqlalchemy.Column(sqlalchemy.String(200), index=True)
228     name = sqlalchemy.Column(sqlalchemy.String(128), index=True)
229     host = sqlalchemy.Column(sqlalchemy.String(128))
230     parameters = sqlalchemy.orm.relationship(
231         'TaskParameter',
232         collection_class=sqlalchemy.orm.collections.attribute_mapped_collection('name'),
233         cascade="all, delete-orphan")
234     events = sqlalchemy.orm.relationship(
235         'TaskEvent',
236         order_by=(sqlalchemy.desc(TaskEvent.ts), sqlalchemy.desc(TaskEvent.id)),
237         backref='task')
238 
239     def __repr__(self):
240         return "TaskRecord(name=%s, host=%s)" % (self.name, self.host)
241 
242 
243 def _upgrade_schema(engine):
244     """
245     Ensure the database schema is up to date with the codebase.
246 
247     :param engine: SQLAlchemy engine of the underlying database.
248     """
249     inspector = reflection.Inspector.from_engine(engine)
250     conn = engine.connect()
251 
252     # Upgrade 1.  Add task_id column and index to tasks
253     if 'task_id' not in [x['name'] for x in inspector.get_columns('tasks')]:
254         logger.warn('Upgrading DbTaskHistory schema: Adding tasks.task_id')
255         conn.execute('ALTER TABLE tasks ADD COLUMN task_id VARCHAR(200)')
256         conn.execute('CREATE INDEX ix_task_id ON tasks (task_id)')
257
```
--- 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/luigi/db_task_history.py b/luigi/db_task_history.py
--- a/luigi/db_task_history.py
+++ b/luigi/db_task_history.py
@@ -196,7 +196,7 @@
     __tablename__ = 'task_parameters'
     task_id = sqlalchemy.Column(sqlalchemy.Integer, sqlalchemy.ForeignKey('tasks.id'), primary_key=True)
     name = sqlalchemy.Column(sqlalchemy.String(128), primary_key=True)
-    value = sqlalchemy.Column(sqlalchemy.String(256))
+    value = sqlalchemy.Column(sqlalchemy.Text())
 
     def __repr__(self):
         return "TaskParameter(task_id=%d, name=%s, value=%s)" % (self.task_id, self.name, self.value)
@@ -247,10 +247,35 @@
     :param engine: SQLAlchemy engine of the underlying database.
     """
     inspector = reflection.Inspector.from_engine(engine)
-    conn = engine.connect()
-
-    # Upgrade 1.  Add task_id column and index to tasks
-    if 'task_id' not in [x['name'] for x in inspector.get_columns('tasks')]:
-        logger.warn('Upgrading DbTaskHistory schema: Adding tasks.task_id')
-        conn.execute('ALTER TABLE tasks ADD COLUMN task_id VARCHAR(200)')
-        conn.execute('CREATE INDEX ix_task_id ON tasks (task_id)')
+    with engine.connect() as conn:
+
+        # Upgrade 1.  Add task_id column and index to tasks
+        if 'task_id' not in [x['name'] for x in inspector.get_columns('tasks')]:
+            logger.warning('Upgrading DbTaskHistory schema: Adding tasks.task_id')
+            conn.execute('ALTER TABLE tasks ADD COLUMN task_id VARCHAR(200)')
+            conn.execute('CREATE INDEX ix_task_id ON tasks (task_id)')
+
+        # Upgrade 2. Alter value column to be TEXT, note that this is idempotent so no if-guard
+        if 'mysql' in engine.dialect.name:
+            conn.execute('ALTER TABLE task_parameters MODIFY COLUMN value TEXT')
+        elif 'oracle' in engine.dialect.name:
+            conn.execute('ALTER TABLE task_parameters MODIFY value TEXT')
+        elif 'mssql' in engine.dialect.name:
+            conn.execute('ALTER TABLE task_parameters ALTER COLUMN value TEXT')
+        elif 'postgresql' in engine.dialect.name:
+            conn.execute('ALTER TABLE task_parameters ALTER COLUMN value TYPE TEXT')
+        elif 'sqlite' in engine.dialect.name:
+            # SQLite does not support changing column types. A database file will need
+            # to be used to pickup this migration change.
+            for i in conn.execute('PRAGMA table_info(task_parameters);').fetchall():
+                if i['name'] == 'value' and i['type'] != 'TEXT':
+                    logger.warning(
+                        'SQLite can not change column types. Please use a new database '
+                        'to pickup column type changes.'
+                    )
+        else:
+            logger.warning(
+                'SQLAlcheny dialect {} could not be migrated to the TEXT type'.format(
+                    engine.dialect
+                )
+            )

{"golden_diff": "diff --git a/luigi/db_task_history.py b/luigi/db_task_history.py\n--- a/luigi/db_task_history.py\n+++ b/luigi/db_task_history.py\n@@ -196,7 +196,7 @@\n     __tablename__ = 'task_parameters'\n     task_id = sqlalchemy.Column(sqlalchemy.Integer, sqlalchemy.ForeignKey('tasks.id'), primary_key=True)\n     name = sqlalchemy.Column(sqlalchemy.String(128), primary_key=True)\n-    value = sqlalchemy.Column(sqlalchemy.String(256))\n+    value = sqlalchemy.Column(sqlalchemy.Text())\n \n     def __repr__(self):\n         return \"TaskParameter(task_id=%d, name=%s, value=%s)\" % (self.task_id, self.name, self.value)\n@@ -247,10 +247,35 @@\n     :param engine: SQLAlchemy engine of the underlying database.\n     \"\"\"\n     inspector = reflection.Inspector.from_engine(engine)\n-    conn = engine.connect()\n-\n-    # Upgrade 1.  Add task_id column and index to tasks\n-    if 'task_id' not in [x['name'] for x in inspector.get_columns('tasks')]:\n-        logger.warn('Upgrading DbTaskHistory schema: Adding tasks.task_id')\n-        conn.execute('ALTER TABLE tasks ADD COLUMN task_id VARCHAR(200)')\n-        conn.execute('CREATE INDEX ix_task_id ON tasks (task_id)')\n+    with engine.connect() as conn:\n+\n+        # Upgrade 1.  Add task_id column and index to tasks\n+        if 'task_id' not in [x['name'] for x in inspector.get_columns('tasks')]:\n+            logger.warning('Upgrading DbTaskHistory schema: Adding tasks.task_id')\n+            conn.execute('ALTER TABLE tasks ADD COLUMN task_id VARCHAR(200)')\n+            conn.execute('CREATE INDEX ix_task_id ON tasks (task_id)')\n+\n+        # Upgrade 2. Alter value column to be TEXT, note that this is idempotent so no if-guard\n+        if 'mysql' in engine.dialect.name:\n+            conn.execute('ALTER TABLE task_parameters MODIFY COLUMN value TEXT')\n+        elif 'oracle' in engine.dialect.name:\n+            conn.execute('ALTER TABLE task_parameters MODIFY value TEXT')\n+        elif 'mssql' in engine.dialect.name:\n+            conn.execute('ALTER TABLE task_parameters ALTER COLUMN value TEXT')\n+        elif 'postgresql' in engine.dialect.name:\n+            conn.execute('ALTER TABLE task_parameters ALTER COLUMN value TYPE TEXT')\n+        elif 'sqlite' in engine.dialect.name:\n+            # SQLite does not support changing column types. A database file will need\n+            # to be used to pickup this migration change.\n+            for i in conn.execute('PRAGMA table_info(task_parameters);').fetchall():\n+                if i['name'] == 'value' and i['type'] != 'TEXT':\n+                    logger.warning(\n+                        'SQLite can not change column types. Please use a new database '\n+                        'to pickup column type changes.'\n+                    )\n+        else:\n+            logger.warning(\n+                'SQLAlcheny dialect {} could not be migrated to the TEXT type'.format(\n+                    engine.dialect\n+                )\n+            )\n", "issue": "Lets use Text type for value column of TaskParameter table\n## Description\n\nChange `value` column type from `string(256)` to `text` because of the issues on insert to task parameters table.\n\n[error](https://gist.github.com/oivoodoo/0fb78f5a041bb08939f30f4af10c00c6) \n## Motivation and Context\n\nFound this issue from the log file in our production system.\n## Have you tested this? If so, how?\n\nGoing to test on 17.10.2016 in the next our deploy.\n\nluigi parameter value column is set to limit size of 256\n<!---\r\nWe use GitHub issues mainly for tracking bugs and feature requests.\r\nQuestions for how to use luigi can be sent to the mailing list.\r\n\r\nCurrently, there are no strict procedures or guidelines for submitting issues.\r\nIn short, please just use common sense.\r\n\r\nCommon sense includes this at bare-minimum:\r\n\r\n * search for similar issues posted before creating a new issue.\r\n * Use markdown to format all code/logs. Issues which are hard to read\r\n   when rendered on GitHub might be closed with a friendly reminder of this.\r\n * If applicable, reading relevant parts of the documentation.\r\n\r\nAlso, add steps to reproduce the bug, if applicable. Sample code would be nice too :)\r\n\r\nFor more information on how to submit valuable contributions,\r\nsee https://opensource.guide/how-to-contribute/#how-to-submit-a-contribution\r\n-->\r\n\r\nBut we need values longer than that. Can the limit be raised? In postgres this can just be `TEXT` type.\n", "before_files": [{"content": "# -*- coding: utf-8 -*-\n#\n# Copyright 2012-2015 Spotify AB\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# 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, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n#\n\n\"\"\"\nProvides a database backend to the central scheduler. This lets you see historical runs.\nSee :ref:`TaskHistory` for information about how to turn out the task history feature.\n\"\"\"\n#\n# Description: Added codes for visualization of how long each task takes\n# running-time until it reaches the next status (failed or done)\n# At \"{base_url}/tasklist\", all completed(failed or done) tasks are shown.\n# At \"{base_url}/tasklist\", a user can select one specific task to see\n# how its running-time has changed over time.\n# At \"{base_url}/tasklist/{task_name}\", it visualizes a multi-bar graph\n# that represents the changes of the running-time for a selected task\n# up to the next status (failed or done).\n# This visualization let us know how the running-time of the specific task\n# has changed over time.\n#\n# Copyright 2015 Naver Corp.\n# Author Yeseul Park (yeseul.park@navercorp.com)\n#\n\nimport datetime\nimport logging\nfrom contextlib import contextmanager\n\nfrom luigi import six\n\nfrom luigi import configuration\nfrom luigi import task_history\nfrom luigi.task_status import DONE, FAILED, PENDING, RUNNING\n\nimport sqlalchemy\nimport sqlalchemy.ext.declarative\nimport sqlalchemy.orm\nimport sqlalchemy.orm.collections\nfrom sqlalchemy.engine import reflection\nBase = sqlalchemy.ext.declarative.declarative_base()\n\nlogger = logging.getLogger('luigi-interface')\n\n\nclass DbTaskHistory(task_history.TaskHistory):\n    \"\"\"\n    Task History that writes to a database using sqlalchemy.\n    Also has methods for useful db queries.\n    \"\"\"\n    CURRENT_SOURCE_VERSION = 1\n\n    @contextmanager\n    def _session(self, session=None):\n        if session:\n            yield session\n        else:\n            session = self.session_factory()\n            try:\n                yield session\n            except BaseException:\n                session.rollback()\n                raise\n            else:\n                session.commit()\n\n    def __init__(self):\n        config = configuration.get_config()\n        connection_string = config.get('task_history', 'db_connection')\n        self.engine = sqlalchemy.create_engine(connection_string)\n        self.session_factory = sqlalchemy.orm.sessionmaker(bind=self.engine, expire_on_commit=False)\n        Base.metadata.create_all(self.engine)\n        self.tasks = {}  # task_id -> TaskRecord\n\n        _upgrade_schema(self.engine)\n\n    def task_scheduled(self, task):\n        htask = self._get_task(task, status=PENDING)\n        self._add_task_event(htask, TaskEvent(event_name=PENDING, ts=datetime.datetime.now()))\n\n    def task_finished(self, task, successful):\n        event_name = DONE if successful else FAILED\n        htask = self._get_task(task, status=event_name)\n        self._add_task_event(htask, TaskEvent(event_name=event_name, ts=datetime.datetime.now()))\n\n    def task_started(self, task, worker_host):\n        htask = self._get_task(task, status=RUNNING, host=worker_host)\n        self._add_task_event(htask, TaskEvent(event_name=RUNNING, ts=datetime.datetime.now()))\n\n    def _get_task(self, task, status, host=None):\n        if task.id in self.tasks:\n            htask = self.tasks[task.id]\n            htask.status = status\n            if host:\n                htask.host = host\n        else:\n            htask = self.tasks[task.id] = task_history.StoredTask(task, status, host)\n        return htask\n\n    def _add_task_event(self, task, event):\n        for (task_record, session) in self._find_or_create_task(task):\n            task_record.events.append(event)\n\n    def _find_or_create_task(self, task):\n        with self._session() as session:\n            if task.record_id is not None:\n                logger.debug(\"Finding task with record_id [%d]\", task.record_id)\n                task_record = session.query(TaskRecord).get(task.record_id)\n                if not task_record:\n                    raise Exception(\"Task with record_id, but no matching Task record!\")\n                yield (task_record, session)\n            else:\n                task_record = TaskRecord(task_id=task._task.id, name=task.task_family, host=task.host)\n                for (k, v) in six.iteritems(task.parameters):\n                    task_record.parameters[k] = TaskParameter(name=k, value=v)\n                session.add(task_record)\n                yield (task_record, session)\n            if task.host:\n                task_record.host = task.host\n        task.record_id = task_record.id\n\n    def find_all_by_parameters(self, task_name, session=None, **task_params):\n        \"\"\"\n        Find tasks with the given task_name and the same parameters as the kwargs.\n        \"\"\"\n        with self._session(session) as session:\n            query = session.query(TaskRecord).join(TaskEvent).filter(TaskRecord.name == task_name)\n            for (k, v) in six.iteritems(task_params):\n                alias = sqlalchemy.orm.aliased(TaskParameter)\n                query = query.join(alias).filter(alias.name == k, alias.value == v)\n\n            tasks = query.order_by(TaskEvent.ts)\n            for task in tasks:\n                # Sanity check\n                assert all(k in task.parameters and v == str(task.parameters[k].value) for (k, v) in six.iteritems(task_params))\n\n                yield task\n\n    def find_all_by_name(self, task_name, session=None):\n        \"\"\"\n        Find all tasks with the given task_name.\n        \"\"\"\n        return self.find_all_by_parameters(task_name, session)\n\n    def find_latest_runs(self, session=None):\n        \"\"\"\n        Return tasks that have been updated in the past 24 hours.\n        \"\"\"\n        with self._session(session) as session:\n            yesterday = datetime.datetime.now() - datetime.timedelta(days=1)\n            return session.query(TaskRecord).\\\n                join(TaskEvent).\\\n                filter(TaskEvent.ts >= yesterday).\\\n                group_by(TaskRecord.id, TaskEvent.event_name, TaskEvent.ts).\\\n                order_by(TaskEvent.ts.desc()).\\\n                all()\n\n    def find_all_runs(self, session=None):\n        \"\"\"\n        Return all tasks that have been updated.\n        \"\"\"\n        with self._session(session) as session:\n            return session.query(TaskRecord).all()\n\n    def find_all_events(self, session=None):\n        \"\"\"\n        Return all running/failed/done events.\n        \"\"\"\n        with self._session(session) as session:\n            return session.query(TaskEvent).all()\n\n    def find_task_by_id(self, id, session=None):\n        \"\"\"\n        Find task with the given record ID.\n        \"\"\"\n        with self._session(session) as session:\n            return session.query(TaskRecord).get(id)\n\n\nclass TaskParameter(Base):\n    \"\"\"\n    Table to track luigi.Parameter()s of a Task.\n    \"\"\"\n    __tablename__ = 'task_parameters'\n    task_id = sqlalchemy.Column(sqlalchemy.Integer, sqlalchemy.ForeignKey('tasks.id'), primary_key=True)\n    name = sqlalchemy.Column(sqlalchemy.String(128), primary_key=True)\n    value = sqlalchemy.Column(sqlalchemy.String(256))\n\n    def __repr__(self):\n        return \"TaskParameter(task_id=%d, name=%s, value=%s)\" % (self.task_id, self.name, self.value)\n\n\nclass TaskEvent(Base):\n    \"\"\"\n    Table to track when a task is scheduled, starts, finishes, and fails.\n    \"\"\"\n    __tablename__ = 'task_events'\n    id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True)\n    task_id = sqlalchemy.Column(sqlalchemy.Integer, sqlalchemy.ForeignKey('tasks.id'), index=True)\n    event_name = sqlalchemy.Column(sqlalchemy.String(20))\n    ts = sqlalchemy.Column(sqlalchemy.TIMESTAMP, index=True, nullable=False)\n\n    def __repr__(self):\n        return \"TaskEvent(task_id=%s, event_name=%s, ts=%s\" % (self.task_id, self.event_name, self.ts)\n\n\nclass TaskRecord(Base):\n    \"\"\"\n    Base table to track information about a luigi.Task.\n\n    References to other tables are available through task.events, task.parameters, etc.\n    \"\"\"\n    __tablename__ = 'tasks'\n    id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True)\n    task_id = sqlalchemy.Column(sqlalchemy.String(200), index=True)\n    name = sqlalchemy.Column(sqlalchemy.String(128), index=True)\n    host = sqlalchemy.Column(sqlalchemy.String(128))\n    parameters = sqlalchemy.orm.relationship(\n        'TaskParameter',\n        collection_class=sqlalchemy.orm.collections.attribute_mapped_collection('name'),\n        cascade=\"all, delete-orphan\")\n    events = sqlalchemy.orm.relationship(\n        'TaskEvent',\n        order_by=(sqlalchemy.desc(TaskEvent.ts), sqlalchemy.desc(TaskEvent.id)),\n        backref='task')\n\n    def __repr__(self):\n        return \"TaskRecord(name=%s, host=%s)\" % (self.name, self.host)\n\n\ndef _upgrade_schema(engine):\n    \"\"\"\n    Ensure the database schema is up to date with the codebase.\n\n    :param engine: SQLAlchemy engine of the underlying database.\n    \"\"\"\n    inspector = reflection.Inspector.from_engine(engine)\n    conn = engine.connect()\n\n    # Upgrade 1.  Add task_id column and index to tasks\n    if 'task_id' not in [x['name'] for x in inspector.get_columns('tasks')]:\n        logger.warn('Upgrading DbTaskHistory schema: Adding tasks.task_id')\n        conn.execute('ALTER TABLE tasks ADD COLUMN task_id VARCHAR(200)')\n        conn.execute('CREATE INDEX ix_task_id ON tasks (task_id)')\n", "path": "luigi/db_task_history.py"}], "after_files": [{"content": "# -*- coding: utf-8 -*-\n#\n# Copyright 2012-2015 Spotify AB\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# 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, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n#\n\n\"\"\"\nProvides a database backend to the central scheduler. This lets you see historical runs.\nSee :ref:`TaskHistory` for information about how to turn out the task history feature.\n\"\"\"\n#\n# Description: Added codes for visualization of how long each task takes\n# running-time until it reaches the next status (failed or done)\n# At \"{base_url}/tasklist\", all completed(failed or done) tasks are shown.\n# At \"{base_url}/tasklist\", a user can select one specific task to see\n# how its running-time has changed over time.\n# At \"{base_url}/tasklist/{task_name}\", it visualizes a multi-bar graph\n# that represents the changes of the running-time for a selected task\n# up to the next status (failed or done).\n# This visualization let us know how the running-time of the specific task\n# has changed over time.\n#\n# Copyright 2015 Naver Corp.\n# Author Yeseul Park (yeseul.park@navercorp.com)\n#\n\nimport datetime\nimport logging\nfrom contextlib import contextmanager\n\nfrom luigi import six\n\nfrom luigi import configuration\nfrom luigi import task_history\nfrom luigi.task_status import DONE, FAILED, PENDING, RUNNING\n\nimport sqlalchemy\nimport sqlalchemy.ext.declarative\nimport sqlalchemy.orm\nimport sqlalchemy.orm.collections\nfrom sqlalchemy.engine import reflection\nBase = sqlalchemy.ext.declarative.declarative_base()\n\nlogger = logging.getLogger('luigi-interface')\n\n\nclass DbTaskHistory(task_history.TaskHistory):\n    \"\"\"\n    Task History that writes to a database using sqlalchemy.\n    Also has methods for useful db queries.\n    \"\"\"\n    CURRENT_SOURCE_VERSION = 1\n\n    @contextmanager\n    def _session(self, session=None):\n        if session:\n            yield session\n        else:\n            session = self.session_factory()\n            try:\n                yield session\n            except BaseException:\n                session.rollback()\n                raise\n            else:\n                session.commit()\n\n    def __init__(self):\n        config = configuration.get_config()\n        connection_string = config.get('task_history', 'db_connection')\n        self.engine = sqlalchemy.create_engine(connection_string)\n        self.session_factory = sqlalchemy.orm.sessionmaker(bind=self.engine, expire_on_commit=False)\n        Base.metadata.create_all(self.engine)\n        self.tasks = {}  # task_id -> TaskRecord\n\n        _upgrade_schema(self.engine)\n\n    def task_scheduled(self, task):\n        htask = self._get_task(task, status=PENDING)\n        self._add_task_event(htask, TaskEvent(event_name=PENDING, ts=datetime.datetime.now()))\n\n    def task_finished(self, task, successful):\n        event_name = DONE if successful else FAILED\n        htask = self._get_task(task, status=event_name)\n        self._add_task_event(htask, TaskEvent(event_name=event_name, ts=datetime.datetime.now()))\n\n    def task_started(self, task, worker_host):\n        htask = self._get_task(task, status=RUNNING, host=worker_host)\n        self._add_task_event(htask, TaskEvent(event_name=RUNNING, ts=datetime.datetime.now()))\n\n    def _get_task(self, task, status, host=None):\n        if task.id in self.tasks:\n            htask = self.tasks[task.id]\n            htask.status = status\n            if host:\n                htask.host = host\n        else:\n            htask = self.tasks[task.id] = task_history.StoredTask(task, status, host)\n        return htask\n\n    def _add_task_event(self, task, event):\n        for (task_record, session) in self._find_or_create_task(task):\n            task_record.events.append(event)\n\n    def _find_or_create_task(self, task):\n        with self._session() as session:\n            if task.record_id is not None:\n                logger.debug(\"Finding task with record_id [%d]\", task.record_id)\n                task_record = session.query(TaskRecord).get(task.record_id)\n                if not task_record:\n                    raise Exception(\"Task with record_id, but no matching Task record!\")\n                yield (task_record, session)\n            else:\n                task_record = TaskRecord(task_id=task._task.id, name=task.task_family, host=task.host)\n                for (k, v) in six.iteritems(task.parameters):\n                    task_record.parameters[k] = TaskParameter(name=k, value=v)\n                session.add(task_record)\n                yield (task_record, session)\n            if task.host:\n                task_record.host = task.host\n        task.record_id = task_record.id\n\n    def find_all_by_parameters(self, task_name, session=None, **task_params):\n        \"\"\"\n        Find tasks with the given task_name and the same parameters as the kwargs.\n        \"\"\"\n        with self._session(session) as session:\n            query = session.query(TaskRecord).join(TaskEvent).filter(TaskRecord.name == task_name)\n            for (k, v) in six.iteritems(task_params):\n                alias = sqlalchemy.orm.aliased(TaskParameter)\n                query = query.join(alias).filter(alias.name == k, alias.value == v)\n\n            tasks = query.order_by(TaskEvent.ts)\n            for task in tasks:\n                # Sanity check\n                assert all(k in task.parameters and v == str(task.parameters[k].value) for (k, v) in six.iteritems(task_params))\n\n                yield task\n\n    def find_all_by_name(self, task_name, session=None):\n        \"\"\"\n        Find all tasks with the given task_name.\n        \"\"\"\n        return self.find_all_by_parameters(task_name, session)\n\n    def find_latest_runs(self, session=None):\n        \"\"\"\n        Return tasks that have been updated in the past 24 hours.\n        \"\"\"\n        with self._session(session) as session:\n            yesterday = datetime.datetime.now() - datetime.timedelta(days=1)\n            return session.query(TaskRecord).\\\n                join(TaskEvent).\\\n                filter(TaskEvent.ts >= yesterday).\\\n                group_by(TaskRecord.id, TaskEvent.event_name, TaskEvent.ts).\\\n                order_by(TaskEvent.ts.desc()).\\\n                all()\n\n    def find_all_runs(self, session=None):\n        \"\"\"\n        Return all tasks that have been updated.\n        \"\"\"\n        with self._session(session) as session:\n            return session.query(TaskRecord).all()\n\n    def find_all_events(self, session=None):\n        \"\"\"\n        Return all running/failed/done events.\n        \"\"\"\n        with self._session(session) as session:\n            return session.query(TaskEvent).all()\n\n    def find_task_by_id(self, id, session=None):\n        \"\"\"\n        Find task with the given record ID.\n        \"\"\"\n        with self._session(session) as session:\n            return session.query(TaskRecord).get(id)\n\n\nclass TaskParameter(Base):\n    \"\"\"\n    Table to track luigi.Parameter()s of a Task.\n    \"\"\"\n    __tablename__ = 'task_parameters'\n    task_id = sqlalchemy.Column(sqlalchemy.Integer, sqlalchemy.ForeignKey('tasks.id'), primary_key=True)\n    name = sqlalchemy.Column(sqlalchemy.String(128), primary_key=True)\n    value = sqlalchemy.Column(sqlalchemy.Text())\n\n    def __repr__(self):\n        return \"TaskParameter(task_id=%d, name=%s, value=%s)\" % (self.task_id, self.name, self.value)\n\n\nclass TaskEvent(Base):\n    \"\"\"\n    Table to track when a task is scheduled, starts, finishes, and fails.\n    \"\"\"\n    __tablename__ = 'task_events'\n    id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True)\n    task_id = sqlalchemy.Column(sqlalchemy.Integer, sqlalchemy.ForeignKey('tasks.id'), index=True)\n    event_name = sqlalchemy.Column(sqlalchemy.String(20))\n    ts = sqlalchemy.Column(sqlalchemy.TIMESTAMP, index=True, nullable=False)\n\n    def __repr__(self):\n        return \"TaskEvent(task_id=%s, event_name=%s, ts=%s\" % (self.task_id, self.event_name, self.ts)\n\n\nclass TaskRecord(Base):\n    \"\"\"\n    Base table to track information about a luigi.Task.\n\n    References to other tables are available through task.events, task.parameters, etc.\n    \"\"\"\n    __tablename__ = 'tasks'\n    id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True)\n    task_id = sqlalchemy.Column(sqlalchemy.String(200), index=True)\n    name = sqlalchemy.Column(sqlalchemy.String(128), index=True)\n    host = sqlalchemy.Column(sqlalchemy.String(128))\n    parameters = sqlalchemy.orm.relationship(\n        'TaskParameter',\n        collection_class=sqlalchemy.orm.collections.attribute_mapped_collection('name'),\n        cascade=\"all, delete-orphan\")\n    events = sqlalchemy.orm.relationship(\n        'TaskEvent',\n        order_by=(sqlalchemy.desc(TaskEvent.ts), sqlalchemy.desc(TaskEvent.id)),\n        backref='task')\n\n    def __repr__(self):\n        return \"TaskRecord(name=%s, host=%s)\" % (self.name, self.host)\n\n\ndef _upgrade_schema(engine):\n    \"\"\"\n    Ensure the database schema is up to date with the codebase.\n\n    :param engine: SQLAlchemy engine of the underlying database.\n    \"\"\"\n    inspector = reflection.Inspector.from_engine(engine)\n    with engine.connect() as conn:\n\n        # Upgrade 1.  Add task_id column and index to tasks\n        if 'task_id' not in [x['name'] for x in inspector.get_columns('tasks')]:\n            logger.warning('Upgrading DbTaskHistory schema: Adding tasks.task_id')\n            conn.execute('ALTER TABLE tasks ADD COLUMN task_id VARCHAR(200)')\n            conn.execute('CREATE INDEX ix_task_id ON tasks (task_id)')\n\n        # Upgrade 2. Alter value column to be TEXT, note that this is idempotent so no if-guard\n        if 'mysql' in engine.dialect.name:\n            conn.execute('ALTER TABLE task_parameters MODIFY COLUMN value TEXT')\n        elif 'oracle' in engine.dialect.name:\n            conn.execute('ALTER TABLE task_parameters MODIFY value TEXT')\n        elif 'mssql' in engine.dialect.name:\n            conn.execute('ALTER TABLE task_parameters ALTER COLUMN value TEXT')\n        elif 'postgresql' in engine.dialect.name:\n            conn.execute('ALTER TABLE task_parameters ALTER COLUMN value TYPE TEXT')\n        elif 'sqlite' in engine.dialect.name:\n            # SQLite does not support changing column types. A database file will need\n            # to be used to pickup this migration change.\n            for i in conn.execute('PRAGMA table_info(task_parameters);').fetchall():\n                if i['name'] == 'value' and i['type'] != 'TEXT':\n                    logger.warning(\n                        'SQLite can not change column types. Please use a new database '\n                        'to pickup column type changes.'\n                    )\n        else:\n            logger.warning(\n                'SQLAlcheny dialect {} could not be migrated to the TEXT type'.format(\n                    engine.dialect\n                )\n            )\n", "path": "luigi/db_task_history.py"}]}