update type annotation

absbox/__init__.py

@@ -1,4 +1,4 @@
-import sys,warnings
+import sys
 
 if (sys.version_info.major >= 3 and sys.version_info.minor < 10):
     raise ImportError("AbsBox support Python with version 3.10+ only")
@@ -9,8 +9,8 @@
 from absbox.local.china import 信贷ABS, SPV
 from absbox.local.cmp import comp_engines
 from absbox.local.generic import Generic
-from absbox.deal import mkDeal, mkDealsBy, setDealsBy, prodDealsBy
-from absbox.local.analytics import run_yield_table
+from absbox.deal import mkDeal, mkDealsBy, setDealsBy, prodDealsBy, setAssumpsBy, prodAssumpsBy
+from absbox.local.analytics import run_yield_table, flow_by_scenario
 from absbox.validation import *
 from importlib.metadata import version
 

absbox/client.py

@@ -3,13 +3,14 @@
 from schema import Schema, Regex
 from json.decoder import JSONDecodeError
 from dataclasses import dataclass
+from typing import Union
 
 import rich
 from rich.console import Console
 import requests
 from requests.exceptions import ConnectionError, ReadTimeout
 import pandas as pd
-from absbox.validation import isValidUrl
+from absbox.validation import isValidUrl,vStr
 
 from absbox.local.util import mkTag, guess_pool_locale, mapValsBy, guess_pool_flow_header \
                               , _read_cf, _read_asset_pricing, mergeStrWithDict \
@@ -213,7 +214,7 @@ def read_single(pool_resp):
             return result
 
     def runStructs(self, deals, poolAssump=None, nonPoolAssump=None, read=True):
-        assert isinstance(deals, dict), f"Deals should be a dict but got {deals}"
+        assert isinstance(deals, dict), f"Deals should be a dict but got {type(deals)}"
         url = f"{self.url}/{Endpoints.RunMultiDeal.value}" 
         _poolAssump = mkAssumpType(poolAssump) if poolAssump else None 
         _nonPerfAssump = mkNonPerfAssumps({}, nonPoolAssump)
@@ -257,7 +258,7 @@ def readResult(x):
 
     def loginLibrary(self, user, pw, **q):
         deal_library_url = q['deal_library']+f"/{LibraryEndpoints.Token.value}"
-        cred = {"user": user, "password": pw}
+        cred = {"user": vStr(user), "password": pw}
         r = self._send_req(json.dumps(cred), deal_library_url)
         if 'token' in r:
             console.print(f"✅{MsgColor.Success.value} login successfully,{r['msg']}")
@@ -340,14 +341,19 @@ def lookupReader(x):
             console.print(f"❌{MsgColor.Error.value}: Failed to read result with error = {e}")
             return None
 
-    def _send_req(self, _req, _url: str, timeout=10, headers={})->dict:
+    def _send_req(self, _req, _url: str, timeout=10, headers={})-> dict | None:
         '''
             send requests to server, raise error if response is not 200
         '''
         with console.status("") as status:
             try:
                 hdrs = self.hdrs | headers
-                r = self.session.post(_url, data=_req.encode('utf-8'), headers=hdrs, verify=False, timeout=timeout)
+                r = None
+                if self.session:
+                    r = self.session.post(_url, data=_req.encode('utf-8'), headers=hdrs, verify=False, timeout=timeout)
+                else:
+                    console.print(f"❌{MsgColor.Error.value} None type for session")
+                    return None
             except (ConnectionRefusedError, ConnectionError):
                 console.print(f"❌{MsgColor.Error.value} Failed to talk to server {_url}")
                 return None

absbox/local/component.py

@@ -1634,7 +1634,7 @@ def mkPricingAssump(x):
             raise RuntimeError(f"Failed to match pricing assumption: {x}")
 
 
-def readPricingResult(x, locale) -> dict:
+def readPricingResult(x, locale) -> dict | None:
     if x is None:
         return None
     h = None
@@ -1646,8 +1646,7 @@ def readPricingResult(x, locale) -> dict:
     elif tag == "ZSpread":
         h = {"cn": ["静态利差"], "en": ["Z-spread"]}
     else:
-        raise RuntimeError(
-            f"Failed to read princing result: {x} with tag={tag}")
+        raise RuntimeError(f"Failed to read princing result: {x} with tag={tag}")
 
     return pd.DataFrame.from_dict({k: v['contents'] for k, v in x.items()}, orient='index', columns=h[locale]).sort_index()
 
@@ -1830,7 +1829,7 @@ def mkRateAssumption(x):
 
 
 def mkNonPerfAssumps(r, xs:list) -> dict:
-    def translate(y):
+    def translate(y) -> dict:
         match y:
             case ("stop", d):
                 return {"stopRunBy":vDate(d)}

absbox/local/util.py

@@ -19,7 +19,7 @@ def flat(xss) -> list:
     return reduce(lambda xs, ys: xs + ys, xss)
 
 
-def mkTag(x: tuple) -> dict:
+def mkTag(x: tuple|str) -> dict:
     match x:
         case (tagName, tagValue):
             return {"tag": tagName, "contents": tagValue}

absbox/validation.py

@@ -4,7 +4,7 @@
 from urllib.parse import urlparse
 
 
-def isValidUrl(url: str) -> str:
+def isValidUrl(url: str) -> str | None:
     try:
         result = urlparse(url)
         if all([result.scheme, result.netloc]):