We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Streams are still exists if parent integration is deleted
**Describe the bug**
Streams are still exists if parent integration is deleted

**To Reproduce**
Steps to reproduce the behavior, for example:
1. Create a Stream Integration
2. Create a Stream which is based on this integration
3.Delete the integration
4. check db. `select * from stream;`
Note, that created stream is still exist while its parent integration is deleted.

**Expected behavior**
Need to delete all related streams in case of deleting a stream integration


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `mindsdb/interfaces/storage/db.py`
Content:
```
1 import os
2 import json
3 import datetime
4 
5 import numpy as np
6 from sqlalchemy import create_engine, orm, types, UniqueConstraint
7 from sqlalchemy.orm import scoped_session, sessionmaker
8 from sqlalchemy.ext.declarative import declarative_base
9 from sqlalchemy import Column, Integer, String, DateTime, Boolean, Index
10 from sqlalchemy.sql.expression import null
11 from sqlalchemy.sql.schema import ForeignKey
12 
13 if os.environ['MINDSDB_DB_CON'].startswith('sqlite:'):
14     engine = create_engine(os.environ['MINDSDB_DB_CON'], echo=False)
15 else:
16     engine = create_engine(os.environ['MINDSDB_DB_CON'], convert_unicode=True, pool_size=30, max_overflow=200, echo=False)
17 Base = declarative_base()
18 session = scoped_session(sessionmaker(bind=engine, autoflush=True))
19 Base.query = session.query_property()
20 
21 
22 # Source: https://stackoverflow.com/questions/26646362/numpy-array-is-not-json-serializable
23 class NumpyEncoder(json.JSONEncoder):
24     """ Special json encoder for numpy types """
25     def default(self, obj):
26         if isinstance(obj, np.integer):
27             return int(obj)
28         elif isinstance(obj, np.floating):
29             return float(obj)
30         elif isinstance(obj, np.ndarray):
31             return obj.tolist()
32         return json.JSONEncoder.default(self, obj)
33 
34 
35 class Array(types.TypeDecorator):
36     ''' Float Type that replaces commas with  dots on input '''
37     impl = types.String
38 
39     def process_bind_param(self, value, dialect):  # insert
40         if isinstance(value, str):
41             return value
42         elif value is None:
43             return value
44         else:
45             return ',|,|,'.join(value)
46 
47     def process_result_value(self, value, dialect):  # select
48         return value.split(',|,|,') if value is not None else None
49 
50 
51 class Json(types.TypeDecorator):
52     ''' Float Type that replaces commas with  dots on input '''
53     impl = types.String
54 
55     def process_bind_param(self, value, dialect):  # insert
56         return json.dumps(value, cls=NumpyEncoder) if value is not None else None
57 
58     def process_result_value(self, value, dialect):  # select
59         return json.loads(value) if value is not None else None
60 
61 
62 class Semaphor(Base):
63     __tablename__ = 'semaphor'
64 
65     id = Column(Integer, primary_key=True)
66     updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)
67     created_at = Column(DateTime, default=datetime.datetime.now)
68     entity_type = Column('entity_type', String)
69     entity_id = Column('entity_id', Integer)
70     action = Column(String)
71     company_id = Column(Integer)
72     uniq_const = UniqueConstraint('entity_type', 'entity_id')
73 
74 
75 class Datasource(Base):
76     __tablename__ = 'datasource'
77 
78     id = Column(Integer, primary_key=True)
79     updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)
80     created_at = Column(DateTime, default=datetime.datetime.now)
81     name = Column(String)
82     data = Column(String)  # Including, e.g. the query used to create it and even the connection info when there's no integration associated with it -- A JSON
83     creation_info = Column(String)
84     analysis = Column(String)  # A JSON
85     company_id = Column(Integer)
86     mindsdb_version = Column(String)
87     datasources_version = Column(String)
88     integration_id = Column(Integer)
89 
90 
91 class Predictor(Base):
92     __tablename__ = 'predictor'
93 
94     id = Column(Integer, primary_key=True)
95     updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)
96     created_at = Column(DateTime, default=datetime.datetime.now)
97     name = Column(String, unique=True)
98     data = Column(Json)  # A JSON -- should be everything returned by `get_model_data`, I think
99     to_predict = Column(Array)
100     company_id = Column(Integer)
101     mindsdb_version = Column(String)
102     native_version = Column(String)
103     datasource_id = Column(Integer)
104     is_custom = Column(Boolean)     # to del
105     learn_args = Column(Json)
106     update_status = Column(String, default='up_to_date')
107 
108     json_ai = Column(Json, nullable=True)
109     code = Column(String, nullable=True)
110     lightwood_version = Column(String, nullable=True)
111     dtype_dict = Column(Json, nullable=True)
112 
113 
114 class AITable(Base):
115     __tablename__ = 'ai_table'
116     id = Column(Integer, primary_key=True)
117     updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)
118     created_at = Column(DateTime, default=datetime.datetime.now)
119     name = Column(String)
120     integration_name = Column(String)
121     integration_query = Column(String)
122     query_fields = Column(Json)
123     predictor_name = Column(String)
124     predictor_columns = Column(Json)
125     company_id = Column(Integer)
126 
127 
128 class Log(Base):
129     __tablename__ = 'log'
130 
131     id = Column(Integer, primary_key=True)
132     created_at = Column(DateTime, default=datetime.datetime.now)
133     log_type = Column(String)  # log, info, warning, traceback etc
134     source = Column(String)  # file + line
135     company_id = Column(Integer)
136     payload = Column(String)
137     created_at_index = Index("some_index", "created_at_index")
138 
139 
140 class Integration(Base):
141     __tablename__ = 'integration'
142     id = Column(Integer, primary_key=True)
143     updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)
144     created_at = Column(DateTime, default=datetime.datetime.now)
145     name = Column(String, nullable=False, unique=True)
146     data = Column(Json)
147     company_id = Column(Integer)
148 
149 
150 class Stream(Base):
151     __tablename__ = 'stream'
152     id = Column(Integer, primary_key=True)
153     name = Column(String, nullable=False, unique=True)
154     stream_in = Column(String, nullable=False)
155     stream_out = Column(String, nullable=False)
156     anomaly_stream = Column(String)
157     learning_stream = Column(String)
158     integration = Column(String, ForeignKey('integration.name', ondelete='CASCADE'), nullable=True)
159     predictor = Column(String, ForeignKey('predictor.name', ondelete='CASCADE'), nullable=False)
160     company_id = Column(Integer)
161     updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)
162     created_at = Column(DateTime, default=datetime.datetime.now)
163     type = Column(String, default='unknown')
164     connection_info = Column(Json, default={})
165 
166 
167 Base.metadata.create_all(engine)
168 orm.configure_mappers()
169
```

Path: `mindsdb/integrations/base/integration.py`
Content:
```
1 import os
2 from threading import Thread
3 from mindsdb.streams import StreamController
4 
5 from mindsdb.utilities.config import STOP_THREADS_EVENT
6 from mindsdb.utilities.log import log
7 import mindsdb.interfaces.storage.db as db
8 
9 
10 class Integration:
11     def __init__(self, config, name):
12         self.config = config
13         self.name = name
14         self.mindsdb_database = config['api']['mysql']['database']
15         self.company_id = os.environ.get('MINDSDB_COMPANY_ID', None)
16 
17     def setup(self):
18         raise NotImplementedError
19 
20     def _query(self, query, fetch=False):
21         raise NotImplementedError
22 
23     def register_predictors(self, model_data_arr):
24         raise NotImplementedError
25 
26     def unregister_predictor(self, name):
27         raise NotImplementedError
28 
29 
30 class StreamIntegration(Integration):
31     def __init__(self, config, name, control_stream=None):
32         Integration.__init__(self, config, name)
33         self._streams = []
34         self._control_stream = control_stream
35     
36     def setup(self):
37         Thread(target=StreamIntegration._loop, args=(self,)).start()
38 
39     def _loop(self):
40         while not STOP_THREADS_EVENT.wait(1.0):
41             if self._control_stream is not None:
42                 # Create or delete streams based on messages from control_stream
43                 for dct in self._control_stream.read():
44                     if 'action' not in dct:
45                         log.error('INTEGRATION %s: no action value found in control record - %s', self.name, dct)
46                     else:
47                         if dct['action'] == 'create':
48                             for k in ['name', 'predictor', 'stream_in', 'stream_out']:
49                                 if k not in dct:
50                                     # Not all required parameters were provided (i.e. stream will not be created)
51                                     # TODO: what's a good way to notify user about this?
52                                     log.error('INTEGRATION %s: stream creating error. not enough data in control record - %s', self.name, dct)
53                                     break
54                             else:
55                                 log.info('INTEGRATION %s: creating stream %s', self.name, dct['name'])
56                                 if db.session.query(db.Stream).filter_by(name=dct['name'], company_id=self.company_id).first() is None:
57                                     stream = db.Stream(
58                                         company_id=self.company_id,
59                                         name=dct['name'],
60                                         integration=self.name,
61                                         predictor=dct['predictor'],
62                                         stream_in=dct['stream_in'],
63                                         stream_out=dct['stream_out'],
64                                         anomaly_stream=dct.get('anomaly_stream', None),
65                                         learning_stream=dct.get('learning_stream', None)
66                                     )
67                                     db.session.add(stream)
68                                     db.session.commit()
69                                 else:
70                                     log.error('INTEGRATION %s: stream with this name already exists - %s', self.name, dct['name'])
71                         elif dct['action'] == 'delete':
72                             for k in ['name']:
73                                 if k not in dct:
74                                     # Not all required parameters were provided (i.e. stream will not be created)
75                                     # TODO: what's a good way to notify user about this?
76                                     log.error('INTEGRATION %s: unable to delete stream - stream name is not provided', self.name)
77                                     break
78                             else:
79                                 log.error('INTEGRATION %s: deleting stream - %s', self.name, dct['name'])
80                                 db.session.query(db.Stream).filter_by(
81                                     company_id=self.company_id,
82                                     integration=self.name,
83                                     name=dct['name']
84                                 ).delete()
85                                 db.session.commit()
86                         else:
87                             # Bad action value
88                             log.error('INTEGRATION %s: bad action value received - %s', self.name, dct)
89                 
90             stream_db_recs = db.session.query(db.Stream).filter_by(
91                 company_id=self.company_id,
92                 integration=self.name
93             ).all()
94 
95             # Stop streams that weren't found in DB
96             indices_to_delete = []
97             for i, s in enumerate(self._streams):
98                 if s.name not in map(lambda x: x.name, stream_db_recs):
99                     log.info("INTEGRATION %s: stopping stream - %s", self.name, s.name)
100                     indices_to_delete.append(i)
101                     self._streams[i].stop_event.set()
102             self._streams = [s for i, s in enumerate(self._streams) if i not in indices_to_delete]
103 
104             # Start new streams found in DB
105             for s in stream_db_recs:
106                 if s.name not in map(lambda x: x.name, self._streams):
107                     self._streams.append(self._make_stream(s))
108 
109         for s in self._streams:
110             s.stop_event.set()
111 
112     def _make_stream(self, s: db.Stream) -> StreamController:
113         raise NotImplementedError
114 
115     def _query(self, query, fetch=False):
116         pass
117 
118     def register_predictors(self, model_data_arr):
119         pass
120 
121     def unregister_predictor(self, name):
122         pass
123
```
--- 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/mindsdb/integrations/base/integration.py b/mindsdb/integrations/base/integration.py
--- a/mindsdb/integrations/base/integration.py
+++ b/mindsdb/integrations/base/integration.py
@@ -37,6 +37,7 @@
         Thread(target=StreamIntegration._loop, args=(self,)).start()
 
     def _loop(self):
+        log.info("INTEGRATION %s: starting", self.name)
         while not STOP_THREADS_EVENT.wait(1.0):
             if self._control_stream is not None:
                 # Create or delete streams based on messages from control_stream
@@ -104,8 +105,10 @@
             # Start new streams found in DB
             for s in stream_db_recs:
                 if s.name not in map(lambda x: x.name, self._streams):
+                    log.info("INTEGRATION %s: starting stream - %s", self.name, s.name)
                     self._streams.append(self._make_stream(s))
 
+        log.info("INTEGRATION %s: stopping", self.name)
         for s in self._streams:
             s.stop_event.set()
 
diff --git a/mindsdb/interfaces/storage/db.py b/mindsdb/interfaces/storage/db.py
--- a/mindsdb/interfaces/storage/db.py
+++ b/mindsdb/interfaces/storage/db.py
@@ -4,7 +4,7 @@
 
 import numpy as np
 from sqlalchemy import create_engine, orm, types, UniqueConstraint
-from sqlalchemy.orm import scoped_session, sessionmaker
+from sqlalchemy.orm import scoped_session, sessionmaker, relationship
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy import Column, Integer, String, DateTime, Boolean, Index
 from sqlalchemy.sql.expression import null
@@ -109,6 +109,7 @@
     code = Column(String, nullable=True)
     lightwood_version = Column(String, nullable=True)
     dtype_dict = Column(Json, nullable=True)
+    streams = relationship("Stream", cascade="all, delete")
 
 
 class AITable(Base):
@@ -145,6 +146,7 @@
     name = Column(String, nullable=False, unique=True)
     data = Column(Json)
     company_id = Column(Integer)
+    streams = relationship("Stream", cascade="all, delete")
 
 
 class Stream(Base):
@@ -155,8 +157,8 @@
     stream_out = Column(String, nullable=False)
     anomaly_stream = Column(String)
     learning_stream = Column(String)
-    integration = Column(String, ForeignKey('integration.name', ondelete='CASCADE'), nullable=True)
-    predictor = Column(String, ForeignKey('predictor.name', ondelete='CASCADE'), nullable=False)
+    integration = Column(String, ForeignKey(Integration.name))
+    predictor = Column(String, ForeignKey(Predictor.name), nullable=False)
     company_id = Column(Integer)
     updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)
     created_at = Column(DateTime, default=datetime.datetime.now)

{"golden_diff": "diff --git a/mindsdb/integrations/base/integration.py b/mindsdb/integrations/base/integration.py\n--- a/mindsdb/integrations/base/integration.py\n+++ b/mindsdb/integrations/base/integration.py\n@@ -37,6 +37,7 @@\n         Thread(target=StreamIntegration._loop, args=(self,)).start()\n \n     def _loop(self):\n+        log.info(\"INTEGRATION %s: starting\", self.name)\n         while not STOP_THREADS_EVENT.wait(1.0):\n             if self._control_stream is not None:\n                 # Create or delete streams based on messages from control_stream\n@@ -104,8 +105,10 @@\n             # Start new streams found in DB\n             for s in stream_db_recs:\n                 if s.name not in map(lambda x: x.name, self._streams):\n+                    log.info(\"INTEGRATION %s: starting stream - %s\", self.name, s.name)\n                     self._streams.append(self._make_stream(s))\n \n+        log.info(\"INTEGRATION %s: stopping\", self.name)\n         for s in self._streams:\n             s.stop_event.set()\n \ndiff --git a/mindsdb/interfaces/storage/db.py b/mindsdb/interfaces/storage/db.py\n--- a/mindsdb/interfaces/storage/db.py\n+++ b/mindsdb/interfaces/storage/db.py\n@@ -4,7 +4,7 @@\n \n import numpy as np\n from sqlalchemy import create_engine, orm, types, UniqueConstraint\n-from sqlalchemy.orm import scoped_session, sessionmaker\n+from sqlalchemy.orm import scoped_session, sessionmaker, relationship\n from sqlalchemy.ext.declarative import declarative_base\n from sqlalchemy import Column, Integer, String, DateTime, Boolean, Index\n from sqlalchemy.sql.expression import null\n@@ -109,6 +109,7 @@\n     code = Column(String, nullable=True)\n     lightwood_version = Column(String, nullable=True)\n     dtype_dict = Column(Json, nullable=True)\n+    streams = relationship(\"Stream\", cascade=\"all, delete\")\n \n \n class AITable(Base):\n@@ -145,6 +146,7 @@\n     name = Column(String, nullable=False, unique=True)\n     data = Column(Json)\n     company_id = Column(Integer)\n+    streams = relationship(\"Stream\", cascade=\"all, delete\")\n \n \n class Stream(Base):\n@@ -155,8 +157,8 @@\n     stream_out = Column(String, nullable=False)\n     anomaly_stream = Column(String)\n     learning_stream = Column(String)\n-    integration = Column(String, ForeignKey('integration.name', ondelete='CASCADE'), nullable=True)\n-    predictor = Column(String, ForeignKey('predictor.name', ondelete='CASCADE'), nullable=False)\n+    integration = Column(String, ForeignKey(Integration.name))\n+    predictor = Column(String, ForeignKey(Predictor.name), nullable=False)\n     company_id = Column(Integer)\n     updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)\n     created_at = Column(DateTime, default=datetime.datetime.now)\n", "issue": "Streams are still exists if parent integration is deleted\n**Describe the bug**\r\nStreams are still exists if parent integration is deleted\r\n\r\n**To Reproduce**\r\nSteps to reproduce the behavior, for example:\r\n1. Create a Stream Integration\r\n2. Create a Stream which is based on this integration\r\n3.Delete the integration\r\n4. check db. `select * from stream;`\r\nNote, that created stream is still exist while its parent integration is deleted.\r\n\r\n**Expected behavior**\r\nNeed to delete all related streams in case of deleting a stream integration\r\n\n", "before_files": [{"content": "import os\nimport json\nimport datetime\n\nimport numpy as np\nfrom sqlalchemy import create_engine, orm, types, UniqueConstraint\nfrom sqlalchemy.orm import scoped_session, sessionmaker\nfrom sqlalchemy.ext.declarative import declarative_base\nfrom sqlalchemy import Column, Integer, String, DateTime, Boolean, Index\nfrom sqlalchemy.sql.expression import null\nfrom sqlalchemy.sql.schema import ForeignKey\n\nif os.environ['MINDSDB_DB_CON'].startswith('sqlite:'):\n    engine = create_engine(os.environ['MINDSDB_DB_CON'], echo=False)\nelse:\n    engine = create_engine(os.environ['MINDSDB_DB_CON'], convert_unicode=True, pool_size=30, max_overflow=200, echo=False)\nBase = declarative_base()\nsession = scoped_session(sessionmaker(bind=engine, autoflush=True))\nBase.query = session.query_property()\n\n\n# Source: https://stackoverflow.com/questions/26646362/numpy-array-is-not-json-serializable\nclass NumpyEncoder(json.JSONEncoder):\n    \"\"\" Special json encoder for numpy types \"\"\"\n    def default(self, obj):\n        if isinstance(obj, np.integer):\n            return int(obj)\n        elif isinstance(obj, np.floating):\n            return float(obj)\n        elif isinstance(obj, np.ndarray):\n            return obj.tolist()\n        return json.JSONEncoder.default(self, obj)\n\n\nclass Array(types.TypeDecorator):\n    ''' Float Type that replaces commas with  dots on input '''\n    impl = types.String\n\n    def process_bind_param(self, value, dialect):  # insert\n        if isinstance(value, str):\n            return value\n        elif value is None:\n            return value\n        else:\n            return ',|,|,'.join(value)\n\n    def process_result_value(self, value, dialect):  # select\n        return value.split(',|,|,') if value is not None else None\n\n\nclass Json(types.TypeDecorator):\n    ''' Float Type that replaces commas with  dots on input '''\n    impl = types.String\n\n    def process_bind_param(self, value, dialect):  # insert\n        return json.dumps(value, cls=NumpyEncoder) if value is not None else None\n\n    def process_result_value(self, value, dialect):  # select\n        return json.loads(value) if value is not None else None\n\n\nclass Semaphor(Base):\n    __tablename__ = 'semaphor'\n\n    id = Column(Integer, primary_key=True)\n    updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)\n    created_at = Column(DateTime, default=datetime.datetime.now)\n    entity_type = Column('entity_type', String)\n    entity_id = Column('entity_id', Integer)\n    action = Column(String)\n    company_id = Column(Integer)\n    uniq_const = UniqueConstraint('entity_type', 'entity_id')\n\n\nclass Datasource(Base):\n    __tablename__ = 'datasource'\n\n    id = Column(Integer, primary_key=True)\n    updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)\n    created_at = Column(DateTime, default=datetime.datetime.now)\n    name = Column(String)\n    data = Column(String)  # Including, e.g. the query used to create it and even the connection info when there's no integration associated with it -- A JSON\n    creation_info = Column(String)\n    analysis = Column(String)  # A JSON\n    company_id = Column(Integer)\n    mindsdb_version = Column(String)\n    datasources_version = Column(String)\n    integration_id = Column(Integer)\n\n\nclass Predictor(Base):\n    __tablename__ = 'predictor'\n\n    id = Column(Integer, primary_key=True)\n    updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)\n    created_at = Column(DateTime, default=datetime.datetime.now)\n    name = Column(String, unique=True)\n    data = Column(Json)  # A JSON -- should be everything returned by `get_model_data`, I think\n    to_predict = Column(Array)\n    company_id = Column(Integer)\n    mindsdb_version = Column(String)\n    native_version = Column(String)\n    datasource_id = Column(Integer)\n    is_custom = Column(Boolean)     # to del\n    learn_args = Column(Json)\n    update_status = Column(String, default='up_to_date')\n\n    json_ai = Column(Json, nullable=True)\n    code = Column(String, nullable=True)\n    lightwood_version = Column(String, nullable=True)\n    dtype_dict = Column(Json, nullable=True)\n\n\nclass AITable(Base):\n    __tablename__ = 'ai_table'\n    id = Column(Integer, primary_key=True)\n    updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)\n    created_at = Column(DateTime, default=datetime.datetime.now)\n    name = Column(String)\n    integration_name = Column(String)\n    integration_query = Column(String)\n    query_fields = Column(Json)\n    predictor_name = Column(String)\n    predictor_columns = Column(Json)\n    company_id = Column(Integer)\n\n\nclass Log(Base):\n    __tablename__ = 'log'\n\n    id = Column(Integer, primary_key=True)\n    created_at = Column(DateTime, default=datetime.datetime.now)\n    log_type = Column(String)  # log, info, warning, traceback etc\n    source = Column(String)  # file + line\n    company_id = Column(Integer)\n    payload = Column(String)\n    created_at_index = Index(\"some_index\", \"created_at_index\")\n\n\nclass Integration(Base):\n    __tablename__ = 'integration'\n    id = Column(Integer, primary_key=True)\n    updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)\n    created_at = Column(DateTime, default=datetime.datetime.now)\n    name = Column(String, nullable=False, unique=True)\n    data = Column(Json)\n    company_id = Column(Integer)\n\n\nclass Stream(Base):\n    __tablename__ = 'stream'\n    id = Column(Integer, primary_key=True)\n    name = Column(String, nullable=False, unique=True)\n    stream_in = Column(String, nullable=False)\n    stream_out = Column(String, nullable=False)\n    anomaly_stream = Column(String)\n    learning_stream = Column(String)\n    integration = Column(String, ForeignKey('integration.name', ondelete='CASCADE'), nullable=True)\n    predictor = Column(String, ForeignKey('predictor.name', ondelete='CASCADE'), nullable=False)\n    company_id = Column(Integer)\n    updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)\n    created_at = Column(DateTime, default=datetime.datetime.now)\n    type = Column(String, default='unknown')\n    connection_info = Column(Json, default={})\n\n\nBase.metadata.create_all(engine)\norm.configure_mappers()\n", "path": "mindsdb/interfaces/storage/db.py"}, {"content": "import os\nfrom threading import Thread\nfrom mindsdb.streams import StreamController\n\nfrom mindsdb.utilities.config import STOP_THREADS_EVENT\nfrom mindsdb.utilities.log import log\nimport mindsdb.interfaces.storage.db as db\n\n\nclass Integration:\n    def __init__(self, config, name):\n        self.config = config\n        self.name = name\n        self.mindsdb_database = config['api']['mysql']['database']\n        self.company_id = os.environ.get('MINDSDB_COMPANY_ID', None)\n\n    def setup(self):\n        raise NotImplementedError\n\n    def _query(self, query, fetch=False):\n        raise NotImplementedError\n\n    def register_predictors(self, model_data_arr):\n        raise NotImplementedError\n\n    def unregister_predictor(self, name):\n        raise NotImplementedError\n\n\nclass StreamIntegration(Integration):\n    def __init__(self, config, name, control_stream=None):\n        Integration.__init__(self, config, name)\n        self._streams = []\n        self._control_stream = control_stream\n    \n    def setup(self):\n        Thread(target=StreamIntegration._loop, args=(self,)).start()\n\n    def _loop(self):\n        while not STOP_THREADS_EVENT.wait(1.0):\n            if self._control_stream is not None:\n                # Create or delete streams based on messages from control_stream\n                for dct in self._control_stream.read():\n                    if 'action' not in dct:\n                        log.error('INTEGRATION %s: no action value found in control record - %s', self.name, dct)\n                    else:\n                        if dct['action'] == 'create':\n                            for k in ['name', 'predictor', 'stream_in', 'stream_out']:\n                                if k not in dct:\n                                    # Not all required parameters were provided (i.e. stream will not be created)\n                                    # TODO: what's a good way to notify user about this?\n                                    log.error('INTEGRATION %s: stream creating error. not enough data in control record - %s', self.name, dct)\n                                    break\n                            else:\n                                log.info('INTEGRATION %s: creating stream %s', self.name, dct['name'])\n                                if db.session.query(db.Stream).filter_by(name=dct['name'], company_id=self.company_id).first() is None:\n                                    stream = db.Stream(\n                                        company_id=self.company_id,\n                                        name=dct['name'],\n                                        integration=self.name,\n                                        predictor=dct['predictor'],\n                                        stream_in=dct['stream_in'],\n                                        stream_out=dct['stream_out'],\n                                        anomaly_stream=dct.get('anomaly_stream', None),\n                                        learning_stream=dct.get('learning_stream', None)\n                                    )\n                                    db.session.add(stream)\n                                    db.session.commit()\n                                else:\n                                    log.error('INTEGRATION %s: stream with this name already exists - %s', self.name, dct['name'])\n                        elif dct['action'] == 'delete':\n                            for k in ['name']:\n                                if k not in dct:\n                                    # Not all required parameters were provided (i.e. stream will not be created)\n                                    # TODO: what's a good way to notify user about this?\n                                    log.error('INTEGRATION %s: unable to delete stream - stream name is not provided', self.name)\n                                    break\n                            else:\n                                log.error('INTEGRATION %s: deleting stream - %s', self.name, dct['name'])\n                                db.session.query(db.Stream).filter_by(\n                                    company_id=self.company_id,\n                                    integration=self.name,\n                                    name=dct['name']\n                                ).delete()\n                                db.session.commit()\n                        else:\n                            # Bad action value\n                            log.error('INTEGRATION %s: bad action value received - %s', self.name, dct)\n                \n            stream_db_recs = db.session.query(db.Stream).filter_by(\n                company_id=self.company_id,\n                integration=self.name\n            ).all()\n\n            # Stop streams that weren't found in DB\n            indices_to_delete = []\n            for i, s in enumerate(self._streams):\n                if s.name not in map(lambda x: x.name, stream_db_recs):\n                    log.info(\"INTEGRATION %s: stopping stream - %s\", self.name, s.name)\n                    indices_to_delete.append(i)\n                    self._streams[i].stop_event.set()\n            self._streams = [s for i, s in enumerate(self._streams) if i not in indices_to_delete]\n\n            # Start new streams found in DB\n            for s in stream_db_recs:\n                if s.name not in map(lambda x: x.name, self._streams):\n                    self._streams.append(self._make_stream(s))\n\n        for s in self._streams:\n            s.stop_event.set()\n\n    def _make_stream(self, s: db.Stream) -> StreamController:\n        raise NotImplementedError\n\n    def _query(self, query, fetch=False):\n        pass\n\n    def register_predictors(self, model_data_arr):\n        pass\n\n    def unregister_predictor(self, name):\n        pass\n", "path": "mindsdb/integrations/base/integration.py"}], "after_files": [{"content": "import os\nimport json\nimport datetime\n\nimport numpy as np\nfrom sqlalchemy import create_engine, orm, types, UniqueConstraint\nfrom sqlalchemy.orm import scoped_session, sessionmaker, relationship\nfrom sqlalchemy.ext.declarative import declarative_base\nfrom sqlalchemy import Column, Integer, String, DateTime, Boolean, Index\nfrom sqlalchemy.sql.expression import null\nfrom sqlalchemy.sql.schema import ForeignKey\n\nif os.environ['MINDSDB_DB_CON'].startswith('sqlite:'):\n    engine = create_engine(os.environ['MINDSDB_DB_CON'], echo=False)\nelse:\n    engine = create_engine(os.environ['MINDSDB_DB_CON'], convert_unicode=True, pool_size=30, max_overflow=200, echo=False)\nBase = declarative_base()\nsession = scoped_session(sessionmaker(bind=engine, autoflush=True))\nBase.query = session.query_property()\n\n\n# Source: https://stackoverflow.com/questions/26646362/numpy-array-is-not-json-serializable\nclass NumpyEncoder(json.JSONEncoder):\n    \"\"\" Special json encoder for numpy types \"\"\"\n    def default(self, obj):\n        if isinstance(obj, np.integer):\n            return int(obj)\n        elif isinstance(obj, np.floating):\n            return float(obj)\n        elif isinstance(obj, np.ndarray):\n            return obj.tolist()\n        return json.JSONEncoder.default(self, obj)\n\n\nclass Array(types.TypeDecorator):\n    ''' Float Type that replaces commas with  dots on input '''\n    impl = types.String\n\n    def process_bind_param(self, value, dialect):  # insert\n        if isinstance(value, str):\n            return value\n        elif value is None:\n            return value\n        else:\n            return ',|,|,'.join(value)\n\n    def process_result_value(self, value, dialect):  # select\n        return value.split(',|,|,') if value is not None else None\n\n\nclass Json(types.TypeDecorator):\n    ''' Float Type that replaces commas with  dots on input '''\n    impl = types.String\n\n    def process_bind_param(self, value, dialect):  # insert\n        return json.dumps(value, cls=NumpyEncoder) if value is not None else None\n\n    def process_result_value(self, value, dialect):  # select\n        return json.loads(value) if value is not None else None\n\n\nclass Semaphor(Base):\n    __tablename__ = 'semaphor'\n\n    id = Column(Integer, primary_key=True)\n    updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)\n    created_at = Column(DateTime, default=datetime.datetime.now)\n    entity_type = Column('entity_type', String)\n    entity_id = Column('entity_id', Integer)\n    action = Column(String)\n    company_id = Column(Integer)\n    uniq_const = UniqueConstraint('entity_type', 'entity_id')\n\n\nclass Datasource(Base):\n    __tablename__ = 'datasource'\n\n    id = Column(Integer, primary_key=True)\n    updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)\n    created_at = Column(DateTime, default=datetime.datetime.now)\n    name = Column(String)\n    data = Column(String)  # Including, e.g. the query used to create it and even the connection info when there's no integration associated with it -- A JSON\n    creation_info = Column(String)\n    analysis = Column(String)  # A JSON\n    company_id = Column(Integer)\n    mindsdb_version = Column(String)\n    datasources_version = Column(String)\n    integration_id = Column(Integer)\n\n\nclass Predictor(Base):\n    __tablename__ = 'predictor'\n\n    id = Column(Integer, primary_key=True)\n    updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)\n    created_at = Column(DateTime, default=datetime.datetime.now)\n    name = Column(String, unique=True)\n    data = Column(Json)  # A JSON -- should be everything returned by `get_model_data`, I think\n    to_predict = Column(Array)\n    company_id = Column(Integer)\n    mindsdb_version = Column(String)\n    native_version = Column(String)\n    datasource_id = Column(Integer)\n    is_custom = Column(Boolean)     # to del\n    learn_args = Column(Json)\n    update_status = Column(String, default='up_to_date')\n\n    json_ai = Column(Json, nullable=True)\n    code = Column(String, nullable=True)\n    lightwood_version = Column(String, nullable=True)\n    dtype_dict = Column(Json, nullable=True)\n    streams = relationship(\"Stream\", cascade=\"all, delete\")\n\n\nclass AITable(Base):\n    __tablename__ = 'ai_table'\n    id = Column(Integer, primary_key=True)\n    updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)\n    created_at = Column(DateTime, default=datetime.datetime.now)\n    name = Column(String)\n    integration_name = Column(String)\n    integration_query = Column(String)\n    query_fields = Column(Json)\n    predictor_name = Column(String)\n    predictor_columns = Column(Json)\n    company_id = Column(Integer)\n\n\nclass Log(Base):\n    __tablename__ = 'log'\n\n    id = Column(Integer, primary_key=True)\n    created_at = Column(DateTime, default=datetime.datetime.now)\n    log_type = Column(String)  # log, info, warning, traceback etc\n    source = Column(String)  # file + line\n    company_id = Column(Integer)\n    payload = Column(String)\n    created_at_index = Index(\"some_index\", \"created_at_index\")\n\n\nclass Integration(Base):\n    __tablename__ = 'integration'\n    id = Column(Integer, primary_key=True)\n    updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)\n    created_at = Column(DateTime, default=datetime.datetime.now)\n    name = Column(String, nullable=False, unique=True)\n    data = Column(Json)\n    company_id = Column(Integer)\n    streams = relationship(\"Stream\", cascade=\"all, delete\")\n\n\nclass Stream(Base):\n    __tablename__ = 'stream'\n    id = Column(Integer, primary_key=True)\n    name = Column(String, nullable=False, unique=True)\n    stream_in = Column(String, nullable=False)\n    stream_out = Column(String, nullable=False)\n    anomaly_stream = Column(String)\n    learning_stream = Column(String)\n    integration = Column(String, ForeignKey(Integration.name))\n    predictor = Column(String, ForeignKey(Predictor.name), nullable=False)\n    company_id = Column(Integer)\n    updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)\n    created_at = Column(DateTime, default=datetime.datetime.now)\n    type = Column(String, default='unknown')\n    connection_info = Column(Json, default={})\n\n\nBase.metadata.create_all(engine)\norm.configure_mappers()\n", "path": "mindsdb/interfaces/storage/db.py"}, {"content": "import os\nfrom threading import Thread\nfrom mindsdb.streams import StreamController\n\nfrom mindsdb.utilities.config import STOP_THREADS_EVENT\nfrom mindsdb.utilities.log import log\nimport mindsdb.interfaces.storage.db as db\n\n\nclass Integration:\n    def __init__(self, config, name):\n        self.config = config\n        self.name = name\n        self.mindsdb_database = config['api']['mysql']['database']\n        self.company_id = os.environ.get('MINDSDB_COMPANY_ID', None)\n\n    def setup(self):\n        raise NotImplementedError\n\n    def _query(self, query, fetch=False):\n        raise NotImplementedError\n\n    def register_predictors(self, model_data_arr):\n        raise NotImplementedError\n\n    def unregister_predictor(self, name):\n        raise NotImplementedError\n\n\nclass StreamIntegration(Integration):\n    def __init__(self, config, name, control_stream=None):\n        Integration.__init__(self, config, name)\n        self._streams = []\n        self._control_stream = control_stream\n    \n    def setup(self):\n        Thread(target=StreamIntegration._loop, args=(self,)).start()\n\n    def _loop(self):\n        log.info(\"INTEGRATION %s: starting\", self.name)\n        while not STOP_THREADS_EVENT.wait(1.0):\n            if self._control_stream is not None:\n                # Create or delete streams based on messages from control_stream\n                for dct in self._control_stream.read():\n                    if 'action' not in dct:\n                        log.error('INTEGRATION %s: no action value found in control record - %s', self.name, dct)\n                    else:\n                        if dct['action'] == 'create':\n                            for k in ['name', 'predictor', 'stream_in', 'stream_out']:\n                                if k not in dct:\n                                    # Not all required parameters were provided (i.e. stream will not be created)\n                                    # TODO: what's a good way to notify user about this?\n                                    log.error('INTEGRATION %s: stream creating error. not enough data in control record - %s', self.name, dct)\n                                    break\n                            else:\n                                log.info('INTEGRATION %s: creating stream %s', self.name, dct['name'])\n                                if db.session.query(db.Stream).filter_by(name=dct['name'], company_id=self.company_id).first() is None:\n                                    stream = db.Stream(\n                                        company_id=self.company_id,\n                                        name=dct['name'],\n                                        integration=self.name,\n                                        predictor=dct['predictor'],\n                                        stream_in=dct['stream_in'],\n                                        stream_out=dct['stream_out'],\n                                        anomaly_stream=dct.get('anomaly_stream', None),\n                                        learning_stream=dct.get('learning_stream', None)\n                                    )\n                                    db.session.add(stream)\n                                    db.session.commit()\n                                else:\n                                    log.error('INTEGRATION %s: stream with this name already exists - %s', self.name, dct['name'])\n                        elif dct['action'] == 'delete':\n                            for k in ['name']:\n                                if k not in dct:\n                                    # Not all required parameters were provided (i.e. stream will not be created)\n                                    # TODO: what's a good way to notify user about this?\n                                    log.error('INTEGRATION %s: unable to delete stream - stream name is not provided', self.name)\n                                    break\n                            else:\n                                log.error('INTEGRATION %s: deleting stream - %s', self.name, dct['name'])\n                                db.session.query(db.Stream).filter_by(\n                                    company_id=self.company_id,\n                                    integration=self.name,\n                                    name=dct['name']\n                                ).delete()\n                                db.session.commit()\n                        else:\n                            # Bad action value\n                            log.error('INTEGRATION %s: bad action value received - %s', self.name, dct)\n                \n            stream_db_recs = db.session.query(db.Stream).filter_by(\n                company_id=self.company_id,\n                integration=self.name\n            ).all()\n\n            # Stop streams that weren't found in DB\n            indices_to_delete = []\n            for i, s in enumerate(self._streams):\n                if s.name not in map(lambda x: x.name, stream_db_recs):\n                    log.info(\"INTEGRATION %s: stopping stream - %s\", self.name, s.name)\n                    indices_to_delete.append(i)\n                    self._streams[i].stop_event.set()\n            self._streams = [s for i, s in enumerate(self._streams) if i not in indices_to_delete]\n\n            # Start new streams found in DB\n            for s in stream_db_recs:\n                if s.name not in map(lambda x: x.name, self._streams):\n                    log.info(\"INTEGRATION %s: starting stream - %s\", self.name, s.name)\n                    self._streams.append(self._make_stream(s))\n\n        log.info(\"INTEGRATION %s: stopping\", self.name)\n        for s in self._streams:\n            s.stop_event.set()\n\n    def _make_stream(self, s: db.Stream) -> StreamController:\n        raise NotImplementedError\n\n    def _query(self, query, fetch=False):\n        pass\n\n    def register_predictors(self, model_data_arr):\n        pass\n\n    def unregister_predictor(self, name):\n        pass\n", "path": "mindsdb/integrations/base/integration.py"}]}