""" AUTHOR: Khushal P Soonderji DATE: Create: Monday, 29th Sept., 2025 OBJECTIVE: To have a way to maintain the Socket.IO app's state maintained and passed across various files. REFERENCES: N/A DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append("../wsio") 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 # For asynchronous activities: import asyncio # To work with various datatypes: from typing import List # Debugging: from icecream import IceCreamDebugger # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class AppState: def __init__(self): super().__setattr__("_data", {}) self.script_data = {} self.init_done = False self.connected_clients = {} self.exclusive_lock = asyncio.Semaphore(1) 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}'") # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": state = AppState() print(state.init_done) state.name = "WSIO" print(state.name) state.name = "WSIO 2" print(state.name)