khushalps/api_utils_converse_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1234

Browse Source
This commit is contained in:
Khushal P Soonderji
2025-09-30 14:15:21 +05:30
parent 0a146c2169
commit c21ec7ba42
8 changed files with 326 additions and 1160 deletions
Download Patch File Download Diff File Expand all files Collapse all files
wsio_v2/app.py
+110 -127
View File
@@ -6,16 +6,16 @@
DATE:
Create: Saturday, 18th May, 2022
Update: Thursday, 22nd Aug. 2024
Create: Monday, 29th Sept., 2025
OBJECTIVE:
To provide an easy way to work with '.json' data and files.
To have a centralized Socket.IO app from where several namespaces can be registered. This is kind of like how
you can have one Quart app and register several blueprints.
REFERENCES:
1) https://www.w3schools.com/python/python_json.asp
N/A
DOWNLOADS:
@@ -37,13 +37,21 @@ sys.path.append(".")
sys.path.append("..")
# System-level activities:
import io
import os
# To work with the JSON standard:
import json
# To make HTTP calls:
import httpx
# To work with files:
from utils_v2.system import files
# To work with SocketIO:
import socket
import socketio
# To maintain the app's state:
from wsio_v2.app_state import AppState
from wsio_v2.test.echo import EchoNamespace
# Debugging:
from icecream import IceCreamDebugger
# *****************************************************************************************************************
@@ -53,7 +61,30 @@ from utils_v2.system import files
# *****************************************************************************************************************
# --- Nothing Yet
# A custom class to maintain the app's state:
app_state = AppState()
# Debugging:
app_state.printer = IceCreamDebugger(prefix = "WSIO | ", includeContext = True)
app_state.no_context_printer = IceCreamDebugger(prefix = "WSIO | ", includeContext = False)
# To make API calls:
app_state.http_client = httpx.AsyncClient(
limits = httpx.Limits(
max_connections = 100, # ............ Maximum number of connections allowed in the pool.
max_keepalive_connections = 50, # ... Maximum number of connections that can be kept alive.
),
timeout = httpx.Timeout(
pool = 120.0, # .... Time to wait for a free connection from the pool.
connect = 2.5, # ... Time to wait for establishing a connection to the server.
write = 10.0, # .... Time to wait for sending data.
read = 9.9 # ....... Time to wait for receiving data.
)
)
# General:
app_state.SERVER_HOSTNAME = str(socket.gethostname())
app_state.ALLOWED_ORIGINS = []
# *****************************************************************************************************************
@@ -63,7 +94,15 @@ from utils_v2.system import files
# *****************************************************************************************************************
# --- Nothing Yet
# For SocketIO:
sio = socketio.AsyncServer(
cors_allowed_origins = "*",
async_mode = "asgi"
)
app = socketio.ASGIApp(sio)
# Register the namespaces:
sio.register_namespace(EchoNamespace(namespace = "/test", app_state = app_state))
# *****************************************************************************************************************
@@ -73,121 +112,7 @@ from utils_v2.system import files
# *****************************************************************************************************************
def from_string(json_data):
"""
Decodes a JSON string to a pythonic variable like a dict.
:param json_data: The JSON string to decode.
:return: The decoded pythonic variable.
"""
python_data = json.loads(json_data)
return python_data
# ---------------------------------------------------------------------------------------------------------------------
def to_string(
python_data,
indent = 4,
default = None,
separators = None,
no_space = False
):
"""
Converts the given pythonic data to a JSON string.
:param python_data: The input data like a dict.
:param indent: The tab-width for pretty presentation.
:param default: The function to use on something that cannot be directly parsed into a JSON string.
:param separators: Custom separators to use.
:param no_space: If you want a dense JSON string that saves memory by not using spaces or tabs or line-breaks. Not
good for human readability, very good for saving memory. WARNING: THIS OVERRIDES EVERY OTHER PARAMETER EXCEPT
'default'.
:return: The JSON string representation of the input pythonic data.
"""
if no_space:
json_data = json.dumps(
python_data,
default = default,
separators = (',', ':')
)
else:
json_data = json.dumps(
python_data,
indent = indent,
default = default,
separators = separators
)
return json_data
# ---------------------------------------------------------------------------------------------------------------------
def from_file(file):
"""
Reads a JSON file and returns it as a pythonic variable like a dict.
:param file: The path to the file on the disk or a file held in RAM as a BytesIO object.
:return: The decoded pythonic variable.
"""
if isinstance(file, io.BytesIO):
file.seek(0)
json_data = file.getvalue()
else: json_data = files.read_file(file)
python_data = from_string(json_data)
return python_data
# ---------------------------------------------------------------------------------------------------------------------
def to_file(
file,
python_data,
indent = 4,
default = None,
separators = None,
no_space = False
):
"""
:param file: Either a path to a file on disk, or a buffer in RAM in the form of a BytesIO object.
:param python_data: The pythonic data to be converted to the JSON string.
:param indent: The tab-width for pretty presentation.
:param default: The function to use on something that cannot be directly parsed into a JSON string.
:param separators: Custom separators to use.
:param no_space: If you want a dense JSON string that saves memory by not using spaces or tabs or line-breaks. Not
good for human readability, very good for saving memory. WARNING: THIS OVERRIDES EVERY OTHER PARAMETER EXCEPT
'default'.
:return: True/False if a path was given, else the same BytesIO object with the written JSON data.
"""
json_data = to_string(
python_data,
indent = indent,
default = default,
separators = separators,
no_space = no_space
)
if isinstance(file, io.BytesIO):
file.write(json_data.encode("utf-8"))
file.seek(0)
return file
else:
try:
files.write_file(file, json_data, mode = "w")
return True
except: return False
# --- Nothing Yet
# *****************************************************************************************************************
@@ -199,4 +124,62 @@ def to_file(
if __name__ == "__main__":
pass
app_state.printer("Main.")
# To get args from the terminal:
import argparse
# To run the ASGI:
import uvicorn
from multiprocessing import freeze_support
# Get the config from the command-line:
parser = argparse.ArgumentParser(description = f"SocketIO to serve live market data (and a general passthrough).")
parser.add_argument(
"-w", "--workers",
type = int,
help = "The no. of threads to spin up for this instance!",
default = 2
)
parser.add_argument(
"-a", "--host",
type = str,
help = "The host for the app. e.g.: '0.0.0.0' or '127.0.0.1'.",
default = "0.0.0.0"
)
parser.add_argument(
"-p", "--port",
type = int,
help = "The port no. to bind the app to.",
default = 8080
)
parser.add_argument(
"-s", "--script-id",
type = str,
help = "The id of this script (will affect the loaded config)."
)
parser.add_argument(
"-d", "--debug",
action = "store_true",
help = "Whether, or not, you want to see debugging messages in the terminal.",
default = False
)
args = parser.parse_args()
# Note down the config;
os.environ["SCRIPT_ID"] = args.script_id
os.environ["DEBUG"] = str(args.debug)
# Startup message:
app_state.printer.enable()
app_state.printer(str(args.debug))
if str(args.debug).lower().find("false") >= 0: app_state.printer.disable()
# Run the gateway:
freeze_support()
uvicorn.run(
app = "app:app",
workers = args.workers,
host = args.host,
port = args.port
)
wsio_v2/app_state.py
+32 -260
View File
@@ -10,8 +10,7 @@
OBJECTIVE:
To have a centralized Socket.IO app from where several namespaces can be registered. This is kind of like how
you can have one Quart app and register several blueprints.
To have a way to maintain the Socket.IO app's state maintained and passed across various files.
REFERENCES:
@@ -83,36 +82,7 @@ from icecream import IceCreamDebugger
# *****************************************************************************************************************
# Debugging:
printer = IceCreamDebugger(prefix = "WSIO | ", includeContext = True)
no_context_printer = IceCreamDebugger(prefix = "WSIO | ", includeContext = False)
# To make API calls:
http_client = httpx.AsyncClient(
limits = httpx.Limits(
max_connections = 100, # ............ Maximum number of connections allowed in the pool.
max_keepalive_connections = 50, # ... Maximum number of connections that can be kept alive.
),
timeout = httpx.Timeout(
pool = 120.0, # .... Time to wait for a free connection from the pool.
connect = 2.5, # ... Time to wait for establishing a connection to the server.
write = 10.0, # .... Time to wait for sending data.
read = 9.9 # ....... Time to wait for receiving data.
)
)
# General:
SERVER_HOSTNAME = str(socket.gethostname())
# For SocketIO:
# Namespaces:
NAMESPACE_TICKS = "/ticks"
NAMESPACE_ORDER_UPDATES = "/order-updates"
# Events:
EVENT_CONNECT = "connect"
EVENT_DISCONNECT = "disconnect"
EVENT_ECHO = "echo"
EVENT_TICKS = "ticks"
# --- Nothing Yet
# *****************************************************************************************************************
@@ -122,22 +92,36 @@ EVENT_TICKS = "ticks"
# *****************************************************************************************************************
# For SocketIO:
ALLOWED_ORIGINS = []
sio = socketio.AsyncServer(
cors_allowed_origins = "*",
async_mode = "asgi"
)
app = socketio.ASGIApp(sio)
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** CLASSES ***
# ***** ****
# *****************************************************************************************************************
# A custom class to maintain the app's state:
class AppState:
def __init__(self):
super().__setattr__("_data", {})
self.script_data = {}
self.init_done = False
self.connected_clients = {}
self.exclusive_lock = asyncio.Semaphore(1)
app_state = AppState()
def __getattr__(self, name):
try: return self._data[name]
except KeyError: raise AttributeError(f"'AppState' has no attribute '{name}'")
def __setattr__(self, name, value):
if name.startswith("_"): super().__setattr__(name, value)
else: self._data[name] = value
def __delattr__(self, name):
if name in self._data: del self._data[name]
else: raise AttributeError(f"'AppState' has no attribute '{name}'")
# *****************************************************************************************************************
@@ -147,163 +131,7 @@ app_state = AppState()
# *****************************************************************************************************************
def origin_is_allowed(origin: str) -> bool:
"""
To check if a given origin is in the allowed list.
:param origin: The origin of your request.
:return: True if allowed, else False.
"""
# Start by assuming failure:
is_allowed = False
# Check through all the allowed origins:
for allowed in ALLOWED_ORIGINS:
try:
if regex.match(origin, allowed):
is_allowed = True
break
except Exception as exception:
printer(exception)
# Done here:
return is_allowed
# ---------------------------------------------------------------------------------------------------------------------
async def init(
script_id: str,
debug: bool
):
"""
To initialize all credentials, instances, and connectivity for this whole script.
:param script_id: The id to use to load cred and data from the internal service.
:param debug: Whether, or not, you would like to print the debug messages.
:return: True if initialized successfully, else False.
"""
# Declare the required global variables:
global SCRIPT_DATA
global ALLOWED_ORIGINS
global redis_cache
global kafka_consumer
# Basic stuff:
if debug: printer.enable()
printer("Initializing.")
# ┏┓ • •
# ┃┃┏┓┓┏┓┓┏┓┏
# ┗┛┛ ┗┗┫┗┛┗┛
# ┛
response = await http_client.post(
url = r"https://api.thecaoffice.com/ca/get/title",
headers = {"Origin": "https://thecaoffice.com/"},
data = {
"domainName": "127.0.0.1:1234",
"screenWidth": 1920,
"screenHeight": 1080
}
)
if response.status_code not in [200]:
print("FATAL: ALLOWED ORIGINS NOT FETCHED!")
return False
ALLOWED_ORIGINS = [origin["domain"] for origin in response.json().get("data", {}).get("rs2", [])]
printer(ALLOWED_ORIGINS)
if len(ALLOWED_ORIGINS) < 1:
print("FATAL: ALLOWED ORIGINS IS EMPTY!")
return False
# ┏┓ ┓ ┓ ┳┓
# ┃ ┏┓┏┓┏┫ ┏┓┏┓┏┫ ┃┃┏┓╋┏┓
# ┗┛┛ ┗ ┗┻ ┗┻┛┗┗┻ ┻┛┗┻┗┗┻
# Get the script credentials:
response = await http_client.get(
url = r"https://nexcom.ditscentre.in/internal/cred/get",
headers = {"X-Script-Id": script_id}
)
if response.status_code not in [200]:
print("FATAL: SCRIPT CREDENTIALS LOADING FAILED!")
return False
script_cred = response.json().get("data")
# Get the script data:
response = await http_client.get(
url = r"https://nexcom.ditscentre.in/internal/data/get",
headers = {"X-Script-Id": script_id}
)
if response.status_code not in [200]:
print("FATAL: SCRIPT DATA LOADING FAILED!")
return False
SCRIPT_DATA = response.json().get("data")
# Done with this step:
printer("Cred and Data loaded.")
# ┓┏┓ ┏┓ ┏┓┓•
# ┃┫ ┏┓╋┃┏┏┓ ┃ ┃┓┏┓┏┓╋┏
# ┛┗┛┗┻┛┛┗┗┻ ┗┛┗┗┗ ┛┗┗┛
# Create the consumer that will listen to changes in watchlist:
consumer_creds = script_cred["kafka"]["consumer"]
kafka_consumer = ConsumerKafka(
topic = consumer_creds["topic"],
bootstrap_servers = consumer_creds["config"]["bootstrapServers"],
security_protocol = consumer_creds["config"].get("securityProtocol", "PLAINTEXT"),
ssl_context = get_ssl_context(
ca_file = consumer_creds["config"].get("caFile"),
cert_file = consumer_creds["config"].get("certFile"),
key_file = consumer_creds["config"].get("keyFile"),
),
serializer = JSONSerializer(),
debug = debug
)
if not await kafka_consumer.connect():
print("FATAL: KAFKA CONSUMER NOT CREATED!")
return False
printer("Kafka consumer ready.")
# ┳┓ ┓• ┏┓ ┓
# ┣┫┏┓┏┫┓┏ ━━ ┃ ┏┓┏┣┓┏┓
# ┛┗┗ ┗┻┗┛ ┗┛┗┻┗┛┗┗
redis_cache = AsyncRedisCache(
connection_string = script_cred["redisCache"]["general"]["sentinelJson"],
# connection_string = script_cred["redisCache"]["general"]["connectionString"],
# serializer = JSONSerializer(),
debug = debug,
debug_prefix = "General Cache | "
)
if not await redis_cache.connect():
print("FATAL: REDIS CACHE NOT CREATED!")
return False
printer("Redis cache ready.")
# ┳┓
# ┃┃┏┓┏┓┏┓
# ┻┛┗┛┛┗┗
# If everything went well, we return with success:
printer("Initialization done.")
return True
# ---------------------------------------------------------------------------------------------------------------------
def register_namespaces():
"""
A quick function that registers all the namespaces to the same Socket.IO app.
"""
printer("Namespaces registered.")
# --- Nothing Yet
# *****************************************************************************************************************
@@ -315,65 +143,9 @@ def register_namespaces():
if __name__ == "__main__":
printer("Main.")
# To get args from the terminal:
import argparse
# To run the ASGI:
import uvicorn
from multiprocessing import freeze_support
# Get the config from the command-line:
parser = argparse.ArgumentParser(description = f"SocketIO to serve live market data (and a general passthrough).")
parser.add_argument(
"-w", "--workers",
type = int,
help = "The no. of threads to spin up for this instance!",
default = 2
)
parser.add_argument(
"-a", "--host",
type = str,
help = "The host for the app. e.g.: '0.0.0.0' or '127.0.0.1'.",
default = "0.0.0.0"
)
parser.add_argument(
"-p", "--port",
type = int,
help = "The port no. to bind the app to.",
default = 8080
)
parser.add_argument(
"-s", "--script-id",
type = str,
help = "The id of this script (will affect the loaded config)."
)
parser.add_argument(
"-d", "--debug",
action = "store_true",
help = "Whether, or not, you want to see debugging messages in the terminal.",
default = False
)
args = parser.parse_args()
# Note down the config;
os.environ["SCRIPT_ID"] = args.script_id
os.environ["DEBUG"] = str(args.debug)
# Startup message:
printer.enable()
printer(str(args.debug))
printer.disable()
# Register all namespaces:
register_namespaces()
# Run the gateway:
freeze_support()
uvicorn.run(
app = "tick_out:app",
workers = args.workers,
host = args.host,
port = args.port
)
state = AppState()
print(state.init_done)
state.name = "WSIO"
print(state.name)
state.name = "WSIO 2"
print(state.name)
wsio_v2/finstitutions/trading/tick_out.py
+89 -39
View File
@@ -10,7 +10,7 @@
OBJECTIVE:
A simple namespace to test the Socket.IO app with a simple echo utility.
A namespace to
REFERENCES:
@@ -32,50 +32,18 @@
# To make sibling directories accessible for imports:
import sys
sys.path.append("../wsio")
sys.path.append(".")
sys.path.append("..")
# System-level activities:
import io
import os
import random
# My utils:
from utils_v2.string import json
from utils_v2.string import regex
from utils_v2.system import files
from utils_v2.date_time import date_time
from utils_v2.database.async_mongo_v2 import AsyncMongo
from utils_v2.queue.kafka.controllers.async_kafka import ConsumerKafka, get_ssl_context
from utils_v2.cache.async_redis_cache_v3 import AsyncRedisCache
from utils_v2.serialization.json_serializer import JSONSerializer
# To make HTTP calls:
import httpx
# To work with date and time:
import datetime
import time
# Models:
from models.core.user import CoreUserInfoModel
from models.finstitutions.trading.ticks import TradingTick
# To work with SocketIO:
import socket
import socketio
# To maintain the app's state:
from wsio_v2.app_state import AppState
# For asynchronous activities:
import asyncio
# To work with various datatypes:
from typing import List
# Debugging:
from icecream import IceCreamDebugger
# For pre-modelled responses:
from utils_v2.api.codes import HttpCodes, StatusCodes
from utils_v2.api.response import ResponseModel
# *****************************************************************************************************************
@@ -112,20 +80,102 @@ class EchoNamespace(socketio.AsyncNamespace):
namespace: str,
app_state: AppState,
):
"""
To set up the namespace that will handle tick data.
:param namespace: The string that defines the path of the namespace. E.g.: "/test".
:param app_state: The AppState that defines the state of the app and carries any custom variables that will be
needed across various namespaces. It's a good way to pass instances of database connections and other shared
variables.
"""
super().__init__(namespace)
self.app_state = app_state
self.app_state.printer("Registered!")
async def on_connect(self, sid, environ, *args):
self.app_state.printer("On Connect", sid)
async def on_connect(self, sid, environ, *args) -> bool | None:
"""
The handler that validates and accepts or rejects incoming connection requests.
:param sid: The socketio session id.
:param environ: The environment information.
:param args: The arguments passed to the handler.
:returns: True if the connection was accepted, else False.
"""
self.app_state.printer("Connection Attempt", sid)
async def on_disconnect(self, sid, reason, *args):
"""
The handler that notes down that the user has been disconnected and frees up any relevant resources.
:param sid: The socketio session id.
:param reason: The reason for termination.
:param args: The arguments passed to the handler.
:returns: None.
"""
self.app_state.printer("On Disconnect", sid)
async def on_echo(self, sid, data):
"""
A test event that can be used to see if the namespace is up and running, or not.
:param sid: The socketio session id.
:param data: The data sent by the client.
:returns: None.
"""
self.app_state.printer("Event", sid, data, type(data).__name__)
await self.emit("echo", data, to = sid)
async def on_set_watchlist(self, sid, data) -> dict:
"""
To handle the request to set the watchlist for a particular user.
:param sid: The socketio session id.
:param data: The data sent by the client.
:returns: None.
"""
self.app_state.printer("Setting Watchlist", sid)
# Start by assuming success:
response = ResponseModel(
status_code = StatusCodes.OK,
message = "Watchlist set successfully.",
data = None
)
# Prepare a list of tickers from the request:
tickers_list = []
# Remove the user from all rooms (tickers):
if response.success:
try:
rooms = self.rooms(sid)
for room in rooms:
if room != sid: # ... Don't remove them from their private room.
await self.leave_room(sid, room)
except Exception as e:
self.app_state.printer("EXCEPTION!", e)
response.status_code = StatusCodes.FAILED
response.message = "Failed to reset previous watchlist."
# Now add the user back to just those rooms (tickers) that he has requested:
if response.success:
try:
for room in tickers_list:
await self.enter_room(sid, room)
except Exception as e:
self.app_state.printer("EXCEPTION!", e)
response.status_code = StatusCodes.FAILED
response.message = "Failed to set new watchlist."
# Done here:
response_dict, response_http_code = response.for_quart()
await self.emit("set_watchlist", response_dict, to = sid)
# *****************************************************************************************************************
# ***** ****
wsio_v2/helpers/helpers.py
+8 -176
View File
@@ -6,12 +6,11 @@
DATE:
Create: Monday, 29th Sept., 2025
Create: Tuesday, 30th Sept., 2025
OBJECTIVE:
To have a centralized Socket.IO app from where several namespaces can be registered. This is kind of like how
you can have one Quart app and register several blueprints.
Simple helper functions to work with the Socket.IO server process.
REFERENCES:
@@ -36,49 +35,14 @@ import sys
sys.path.append(".")
sys.path.append("..")
# System-level activities:
import io
import os
import random
# My utils:
from utils_v2.string import json
from utils_v2.string import regex
from utils_v2.system import files
from utils_v2.date_time import date_time
from utils_v2.database.async_mongo_v2 import AsyncMongo
from utils_v2.queue.kafka.controllers.async_kafka import ConsumerKafka, get_ssl_context
from utils_v2.cache.async_redis_cache_v3 import AsyncRedisCache
from utils_v2.serialization.json_serializer import JSONSerializer
# To make HTTP calls:
import httpx
# To work with date and time:
import datetime
import time
# Models:
from models.core.user import CoreUserInfoModel
from models.finstitutions.trading.ticks import TradingTick
# To work with SocketIO:
import socket
import socketio
# To maintain the app's state:
from wsio_v2.app_state import AppState
from wsio_v2.test.test_ns import TestNamespace
from wsio_v2.test.echo import echo
# For asynchronous activities:
import asyncio
# To work with various datatypes:
from typing import List
# Debugging:
from icecream import IceCreamDebugger
# *****************************************************************************************************************
@@ -88,42 +52,7 @@ from icecream import IceCreamDebugger
# *****************************************************************************************************************
# A custom class to maintain the app's state:
app_state = AppState()
# Debugging:
app_state.printer = IceCreamDebugger(prefix = "WSIO | ", includeContext = True)
app_state.no_context_printer = IceCreamDebugger(prefix = "WSIO | ", includeContext = False)
# To make API calls:
app_state.http_client = httpx.AsyncClient(
limits = httpx.Limits(
max_connections = 100, # ............ Maximum number of connections allowed in the pool.
max_keepalive_connections = 50, # ... Maximum number of connections that can be kept alive.
),
timeout = httpx.Timeout(
pool = 120.0, # .... Time to wait for a free connection from the pool.
connect = 2.5, # ... Time to wait for establishing a connection to the server.
write = 10.0, # .... Time to wait for sending data.
read = 9.9 # ....... Time to wait for receiving data.
)
)
# General:
app_state.SERVER_HOSTNAME = str(socket.gethostname())
app_state.ALLOWED_ORIGINS = []
# Namespaces:
app_state.NAMESPACE_DEFAULT = "/"
app_state.NAMESPACE_TEST = "/test"
app_state.NAMESPACE_TICKS = "/ticks"
app_state.NAMESPACE_ORDER_UPDATES = "/order-updates"
# Events:
app_state.EVENT_CONNECT = "connect"
app_state.EVENT_DISCONNECT = "disconnect"
app_state.EVENT_ECHO = "echo"
app_state.EVENT_TICKS = "ticks"
# --- Nothing Yet
# *****************************************************************************************************************
@@ -133,13 +62,7 @@ app_state.EVENT_TICKS = "ticks"
# *****************************************************************************************************************
# For SocketIO:
sio = socketio.AsyncServer(
cors_allowed_origins = "*",
async_mode = "asgi"
)
app = socketio.ASGIApp(sio)
sio.register_namespace(TestNamespace("/test"))
# --- Nothing Yet
# *****************************************************************************************************************
@@ -149,7 +72,7 @@ sio.register_namespace(TestNamespace("/test"))
# *****************************************************************************************************************
def origin_is_allowed(origin: str) -> bool:
def origin_is_allowed(origin: str, app_state: AppState) -> bool:
"""
To check if a given origin is in the allowed list.
@@ -182,7 +105,8 @@ def origin_is_allowed(origin: str) -> bool:
async def init(
script_id: str,
debug: bool
debug: bool,
app_state: AppState,
):
"""
@@ -192,12 +116,6 @@ async def init(
:return: True if initialized successfully, else False.
"""
# Declare the required global variables:
# global SCRIPT_DATA
# global ALLOWED_ORIGINS
# global redis_cache
# global kafka_consumer
# Basic stuff:
if debug: app_state.printer.enable()
app_state.printer("Initializing.")
@@ -300,29 +218,6 @@ async def init(
return True
# ---------------------------------------------------------------------------------------------------------------------
def register_namespaces():
"""
A quick function that registers all the namespaces to the same Socket.IO app.
"""
app_state.printer("Registering namespaces.")
sio.register_namespace(TestNamespace("/test"))
app_state.printer("Namespaces registered.")
# ---------------------------------------------------------------------------------------------------------------------
@sio.on(event = app_state.EVENT_ECHO, namespace = app_state.NAMESPACE_DEFAULT)
async def echo(sid, data):
print(f"Received (default ns) from {sid}: {data}")
await sio.emit("echo", data, to = sid)
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
@@ -332,67 +227,4 @@ async def echo(sid, data):
if __name__ == "__main__":
app_state.printer("Main.")
# To get args from the terminal:
import argparse
# To run the ASGI:
import uvicorn
from multiprocessing import freeze_support
# Get the config from the command-line:
parser = argparse.ArgumentParser(description = f"SocketIO to serve live market data (and a general passthrough).")
parser.add_argument(
"-w", "--workers",
type = int,
help = "The no. of threads to spin up for this instance!",
default = 2
)
parser.add_argument(
"-a", "--host",
type = str,
help = "The host for the app. e.g.: '0.0.0.0' or '127.0.0.1'.",
default = "0.0.0.0"
)
parser.add_argument(
"-p", "--port",
type = int,
help = "The port no. to bind the app to.",
default = 8080
)
parser.add_argument(
"-s", "--script-id",
type = str,
help = "The id of this script (will affect the loaded config)."
)
parser.add_argument(
"-d", "--debug",
action = "store_true",
help = "Whether, or not, you want to see debugging messages in the terminal.",
default = False
)
args = parser.parse_args()
# Note down the config;
os.environ["SCRIPT_ID"] = args.script_id
os.environ["DEBUG"] = str(args.debug)
# Startup message:
app_state.printer.enable()
app_state.printer(str(args.debug))
if str(args.debug).lower().find("false") >= 0: app_state.printer.disable()
# Register all namespaces:
app_state.init_func = init
app_state.origin_check_func = origin_is_allowed
# register_namespaces()
# Run the gateway:
freeze_support()
uvicorn.run(
app = "app:app",
workers = args.workers,
host = args.host,
port = args.port
)
pass
wsio_v2/test/echo.py
+13 -47
View File
@@ -32,51 +32,15 @@
# To make sibling directories accessible for imports:
import sys
sys.path.append("../wsio")
sys.path.append(".")
sys.path.append("..")
# System-level activities:
import io
import os
import random
# My utils:
from utils_v2.string import json
from utils_v2.string import regex
from utils_v2.system import files
from utils_v2.date_time import date_time
from utils_v2.database.async_mongo_v2 import AsyncMongo
from utils_v2.queue.kafka.controllers.async_kafka import ConsumerKafka, get_ssl_context
from utils_v2.cache.async_redis_cache_v3 import AsyncRedisCache
from utils_v2.serialization.json_serializer import JSONSerializer
# To make HTTP calls:
import httpx
# To work with date and time:
import datetime
import time
# Models:
from models.core.user import CoreUserInfoModel
from models.finstitutions.trading.ticks import TradingTick
# To work with SocketIO:
import socket
import socketio
# To maintain the app's state:
from wsio_v2.app_state import AppState
# For asynchronous activities:
import asyncio
# To work with various datatypes:
from typing import List
# Debugging:
from icecream import IceCreamDebugger
# *****************************************************************************************************************
# ***** ****
@@ -105,24 +69,26 @@ from icecream import IceCreamDebugger
# *****************************************************************************************************************
class TestNamespace(socketio.AsyncNamespace):
class EchoNamespace(socketio.AsyncNamespace):
# Internal variables:
app_state: AppState = None
def __init__(self, namespace):
def __init__(
self,
namespace: str,
app_state: AppState,
):
super().__init__(namespace)
print("Test Namespace Init")
self.app_state = app_state
self.app_state.printer("Registered!")
async def on_connect(self, sid, environ, *args):
print(f"ON CONNECT TEST — SID: {sid}")
self.app_state.printer("On Connect", sid)
async def on_disconnect(self, sid, reason, *args):
print(f"ON DISCONNECT TEST — SID: {sid}")
self.app_state.printer("On Disconnect", sid)
async def on_echo(self, sid, data):
print(f"Received echo from {sid}: {data}")
await self.emit("echo", data, to=sid)
self.app_state.printer("Event", sid, data, type(data).__name__)
await self.emit("echo", data, to = sid)
# *****************************************************************************************************************