(20241231) CORS on SocketIO.

api/blueprints/finstitutions/trading/symbols/list.py

@@ -25,7 +25,8 @@
         N/A
 
 """
-import datetime
+
+
 #   *****************************************************************************************************************
 #  *****                                                                                                          ****
 # ***                                                   IMPORT                                                      ***
@@ -69,6 +70,9 @@ from models.api.finstitutions.trading.symbols.list import (
     TradingSymbolListBrokerResponse
 )
 
+# To work with dat and time:
+import datetime
+
 # For asynchronous activities:
 import asyncio
 

background/finstitutions/trading/to_kafka.py

@@ -43,7 +43,7 @@ import os
 # My utils:
 from utils_v2.string import json
 from utils_v2.system import files
-from utils_v2.queue.kafka import ProducerKafka, ConsumerKafka
+from utils_v2.queue.kafka.controllers.kafka import ProducerKafka, ConsumerKafka
 
 # To make HTTP calls:
 import httpx
@@ -134,6 +134,9 @@ def flush_kafka():
     kafka_producer.flush()
 
 
+# ---------------------------------------------------------------------------------------------------------------------
+
+
 def to_kafka(tick: TradingTick) -> bool:
 
     global tick_count

utils_v2/queue/kafka/controllers/kafka.py

@@ -305,7 +305,7 @@ class ConsumerKafka:
     @staticmethod
     def create_config(
             bootstrap_servers: str | List[str],
-            group_id: str,
+            group_id: str = "default",
             auto_offset_reset: Literal["latest", "earliest"] = "latest",
             security_protocol: Literal["PLAINTEXT", "SSL"] = "PLAINTEXT",
             ca_file: str | None = None,

wsio/finstitutions/trading/main.py

@@ -41,6 +41,7 @@ import os
 
 # My utils:
 from utils_v2.string import json
+from utils_v2.string import regex
 from utils_v2.system import files
 from utils_v2.database.async_mongo_v2 import AsyncMongo
 from utils_v2.queue.kafka.controllers.async_kafka import ConsumerKafka, get_ssl_context
@@ -86,11 +87,33 @@ printer.disable()
 SERVER_HOSTNAME = str(socket.gethostname())
 
 # For SocketIO:
-sio = socketio.AsyncServer(async_mode = "asgi")
+# Custom CORS function to allow local IPs
+def allow_origins(origin):
+
+    # Allow specific domains
+    allowed_origins = [
+        r".*\.thecaoffice\.com.*",
+        r".*\.ditscentre\.in.*",
+        r"http[s]?://127\.0\.0\.1.*",
+        r"http[s]?://192\.168\.[\d]{1,3}\.[\d]{1,3}.*",
+    ]
+
+    # Check if the origin is in the allowed list
+    for allowed_origin in allowed_origins:
+        if regex.match(origin, allowed_origin): return True
+
+    # Reject other origins
+    return False
+
+sio = socketio.AsyncServer(
+    cors_allowed_origins = allow_origins,
+    async_mode = "asgi"
+)
 app = socketio.ASGIApp(sio)
 
 # SocketIO Namespaces:
-NAMESPACE_MODULE = "/finstitutions/trading"
+# NAMESPACE_MODULE = "/finstitutions/trading"
+NAMESPACE_MODULE = None
 NAMESPACE_PASSTHROUGH = "/passthrough"
 
 # SocketIO Events:
@@ -147,7 +170,7 @@ async def handle_connect(sid, environ) -> bool:
 
     # Allow/reject requests:
     printer(sid)
-    print("ENVIRON:", json.to_string(environ, default = str))
+    # print("ENVIRON:", json.to_string(environ, default = str))
     while redis_cache is None: await asyncio.sleep(0.5)
     result = await redis_cache.set(
         key = connected_clients[sid]["redisKey"],