cdb: Related data source and oracle script

CHANGELOG_UNRELEASED.md

@@ -16,6 +16,8 @@
 
 ### Emitter & Flusher
 
+ - (impv) [\#2475](https://github.com/bandprotocol/bandchain/pull/2475) Add related data source and oracle script table.
+
 ### Scan
 
 - (feat) [\#2482](https://github.com/bandprotocol/bandchain/pull/2482) Implemented latest requests.

flusher/flusher/db.py

@@ -319,3 +319,10 @@ def Column(*args, **kwargs):
     Column("answer", CustomVoteOption),
     Column("tx_id", sa.Integer, sa.ForeignKey("transactions.id")),
 )
+
+related_data_source_oracle_scripts = sa.Table(
+    "related_data_source_oracle_scripts",
+    metadata,
+    Column("data_source_id", sa.Integer, sa.ForeignKey("data_sources.id"), primary_key=True),
+    Column("oracle_script_id", sa.Integer, sa.ForeignKey("oracle_scripts.id"), primary_key=True),
+)

flusher/flusher/handler.py

@@ -23,6 +23,7 @@
     proposals,
     deposits,
     votes,
+    related_data_source_oracle_scripts,
 )
 
 
@@ -103,7 +104,22 @@ def handle_update_request(self, msg):
             condition = (col == msg[col.name]) & condition
         self.conn.execute(requests.update().where(condition).values(**msg))
 
+    def handle_update_related_ds_os(self, msg):
+        self.conn.execute(
+            insert(related_data_source_oracle_scripts)
+            .values(**msg)
+            .on_conflict_do_nothing(constraint="related_data_source_oracle_scripts_pkey")
+        )
+
     def handle_new_raw_request(self, msg):
+        self.handle_update_related_ds_os(
+            {
+                "oracle_script_id": self.conn.execute(
+                    select([requests.c.oracle_script_id]).where(accounts.c.id == msg["request_id"])
+                ).scalar(),
+                "data_source_id": msg["data_source_id"],
+            }
+        )
         self.conn.execute(raw_requests.insert(), msg)
 
     def handle_new_val_request(self, msg):