diff --git a/README.md b/README.md index 28c775f..03442e4 100644 --- a/README.md +++ b/README.md @@ -22,7 +22,7 @@ Please generously STAR★ our project or donate to us! ## 🚀 Key Features -- **Configuration Management (`aloha.config`)**: Lazy-loaded settings (`SETTINGS`) using HOCON (Human-Optimized Config Object Notation), supporting environment profile overrides (`ENV_PROFILE` / `FILES_CONFIG`) and environment variable injection. +- **Configuration Management (`aloha.config`)**: Lazy-loaded settings (`SETTINGS`) using HOCON (Human-Optimized Config Object Notation), supporting environment profile overrides (`PROFILE_ENV` / `FILES_CONFIG`, with legacy `ENV_PROFILE` deprecated) and environment variable injection. - **Concurrent-Safe Logging (`aloha.logger`)**: Multi-process safe daily rotating log file handler, console output, and automatic log paths configuration. - **Database Operators (`aloha.db`)**: Pre-built SQLAlchemy-backed connections for PostgreSQL, MySQL, SQLite, DuckDB, MongoDB, Redis, Elasticsearch, and Kafka, with password resolution via a secure `PasswordVault` wrapper. - **Encryption & Utilities (`aloha.encrypt`)**: Fast helpers for AES (ECB/CBC) encryption, RSA asymmetric key-pair generation/signatures, JWT encoding/decoding, and Base62 hashing. diff --git a/doc/en/README-config.md b/doc/en/README-config.md index 3c94b20..94e776f 100644 --- a/doc/en/README-config.md +++ b/doc/en/README-config.md @@ -2,14 +2,18 @@ ## OS environment variables -### `ENV_PROFILE` +### `PROFILE_ENV` *Default value*: `None` (not defined). Define the environment profile for the current process, such as `DEV | STG | PRD`. This is usually used to decide which config file in `${DIR_CONFIG}` should be used as the entrypoint config. -If this environment variable is defined, `aloha` will first search for `main-${ENV_PROFILE}.conf`; otherwise it uses `main.conf`. +If this environment variable is defined, `aloha` will first search for `main-${PROFILE_ENV}.conf`; otherwise it uses `main.conf`. + +> [!NOTE] +> **Backward Compatibility & Deprecation Notice**: +> If `PROFILE_ENV` is not defined or has no value, `aloha` falls back to reading the legacy `ENV_PROFILE` environment variable. If `ENV_PROFILE` is present, a `DeprecationWarning` will be emitted. `ENV_PROFILE` is deprecated and support will be removed in a future release. Please migrate to `PROFILE_ENV`. ### `ENTRYPOINT` @@ -49,4 +53,4 @@ Define where to find configuration files. *Default value*: `None` (not defined). Optional. Define a comma-separated list of config files to load. -If this variable is set, `ENV_PROFILE` is ignored. +If this variable is set, `PROFILE_ENV` (and legacy `ENV_PROFILE`) is ignored. diff --git a/doc/en/api/util.md b/doc/en/api/util.md index fc71493..824beb0 100644 --- a/doc/en/api/util.md +++ b/doc/en/api/util.md @@ -16,7 +16,7 @@ ## Time Utilities (`aloha.util.time`) -This module provides tools for wrapping function calls (such as HTTP requests via `requests` or `httpx`) with time constraints (timeouts), allowing execution of optional callbacks upon completion or failure. +This module provides tools for wrapping function calls (such as HTTP requests via `httpx2`) with time constraints (timeouts), allowing execution of optional callbacks upon completion or failure. ### Key Functions - `run_with_timeout`: Wrap a synchronous function call with a timeout. @@ -25,7 +25,7 @@ This module provides tools for wrapping function calls (such as HTTP requests vi ### Usage Example ```python from aloha.util.time import run_with_timeout -import requests +import httpx2 def success_callback(response): print("Request succeeded:", response.status_code) @@ -36,7 +36,7 @@ def fail_callback(exception): # Synchronous call with timeout try: run_with_timeout( - requests.get, + httpx2.get, 2.5, # 2.5 seconds timeout "https://httpbin.org/delay/1", fn_callback_success=success_callback, diff --git a/doc/skills/aloha_python/SKILL.md b/doc/skills/aloha_python/SKILL.md index bf5f95c..ec5c66a 100644 --- a/doc/skills/aloha_python/SKILL.md +++ b/doc/skills/aloha_python/SKILL.md @@ -24,7 +24,7 @@ When developing Python code in this codebase, adhere to the following naming con The `aloha` package is divided into several specialized sub-modules. See their respective reference documents for detailed APIs and usage examples: -- **[Configuration Management (`aloha.config`)](references/config.md)**: HOCON configuration loading, environment profile handling (`ENV_PROFILE`), and settings loading via `SETTINGS`. +- **[Configuration Management (`aloha.config`)](references/config.md)**: HOCON configuration loading, environment profile handling (`PROFILE_ENV`, legacy `ENV_PROFILE` is deprecated), and settings loading via `SETTINGS`. - **[Logging Framework (`aloha.logger`)](references/logger.md)**: Safe concurrent multi-process logging, console handlers, and custom logger setups. - **[Encryption & Hashing (`aloha.encrypt`)](references/encrypt.md)**: AES encryption, RSA keys generation & signing/verifying, JWT encoding/decoding, and dictionary/object hashing helpers. - **[Testing Utilities (`aloha.testing`)](references/testing.md)**: Base `UnitTestCase` class with pre-configured settings/loggers, and `ServiceTestCase` for testing HTTP API endpoints. diff --git a/doc/skills/aloha_python/references/config.md b/doc/skills/aloha_python/references/config.md index 3afe819..4f41ff0 100644 --- a/doc/skills/aloha_python/references/config.md +++ b/doc/skills/aloha_python/references/config.md @@ -10,13 +10,13 @@ The configuration and startup behavior of an `aloha` application can be customiz | Environment Variable | Default Value | Description | | :------------------- | :----------------------- | :------------------------------------------------------------------------------------------------------------------------------- | -| `ENV_PROFILE` | `None` (undefined) | Specifies the running profile (e.g., `DEV`, `STG`, `PRD`). Determines the configuration entry file (`main-${ENV_PROFILE}.conf`). | +| `PROFILE_ENV` | `None` (undefined) | Specifies the running profile (e.g., `DEV`, `STG`, `PRD`). Determines the configuration entry file (`main-${PROFILE_ENV}.conf`). Falls back to legacy `ENV_PROFILE` with a deprecation warning if unset. `ENV_PROFILE` is deprecated and will be removed in a future release. | | `ENTRYPOINT` | `None` (undefined) | Specifies the default Python module entry point when running `aloha start`. The module must contain a `main()` function. | | `APP_MODULE` | `default` | Identifies the application/module name. Mapped to configuration key `APP_MODULE` and used as a prefix for the log file names. | | `DIR_LOG` | `logs` | The directory where log files are stored. | | `DIR_RESOURCE` | `resource` (under CWD) | Root directory containing non-code resources (e.g. assets, static data). | | `DIR_CONFIG` | `${DIR_RESOURCE}/config` | Directory where configuration files are located. | -| `FILES_CONFIG` | `None` (undefined) | Comma-separated list of configuration filenames to load (e.g., `db.conf,server.conf`). If defined, it overrides `ENV_PROFILE`. | +| `FILES_CONFIG` | `None` (undefined) | Comma-separated list of configuration filenames to load (e.g., `db.conf,server.conf`). If defined, it overrides `PROFILE_ENV`. | --- @@ -30,8 +30,10 @@ This module resolves paths for config directories, resource directories, and act - `get_config_dir(*args) -> str`: Resolves the absolute path to the configuration directory. Relies on the `DIR_CONFIG` environment variable. - `get_config_files() -> list`: Determines which HOCON configuration files should be loaded. - If `FILES_CONFIG` environment variable is defined, it splits the list by comma and resolves their paths. - - If `FILES_CONFIG` is not defined but `ENV_PROFILE` is defined, it resolves `main-${ENV_PROFILE}.conf`. - - Otherwise, it defaults to `main.conf`. + - If `FILES_CONFIG` is not defined: + - Checks `PROFILE_ENV` first. If defined, it resolves `main-${PROFILE_ENV}.conf`. + - If `PROFILE_ENV` is not defined, falls back to legacy `ENV_PROFILE` with a `DeprecationWarning` (support for `ENV_PROFILE` will be removed in a future release). + - Otherwise (neither is defined), it defaults to `main.conf`. - `get_project_base_dir(file_caller: str) -> str`: Traverses directories upwards from `file_caller` (typically passed as `__file__`) until it finds a directory containing no `__init__.py`, marking the project base root. --- diff --git a/doc/zh/README-config.md b/doc/zh/README-config.md index 15446ea..73a239e 100644 --- a/doc/zh/README-config.md +++ b/doc/zh/README-config.md @@ -2,14 +2,18 @@ ## 操作系统环境变量 -### `ENV_PROFILE` +### `PROFILE_ENV` *默认值*：`None`（未定义）。 用于指定当前进程运行环境，例如 `DEV | STG | PRD`。 通常用于决定 `${DIR_CONFIG}` 下哪个配置文件作为入口配置。 -如果该变量已定义，`aloha` 会优先查找 `main-${ENV_PROFILE}.conf`；否则使用 `main.conf`。 +如果该变量已定义，`aloha` 会优先查找 `main-${PROFILE_ENV}.conf`；否则使用 `main.conf`。 + +> [!NOTE] +> **向后兼容与弃用说明**： +> 如果 `PROFILE_ENV` 未定义或无值，`aloha` 会回退读取旧版环境变量 `ENV_PROFILE`。若检测到 `ENV_PROFILE` 有值，系统将输出 `DeprecationWarning` 弃用警告。`ENV_PROFILE` 已被弃用，并且在将来版本中会正式取消支持。请尽快迁移使用 `PROFILE_ENV`。 ### `ENTRYPOINT` @@ -49,4 +53,4 @@ *默认值*：`None`（未定义）。 可选项。用于定义以英文逗号分隔的配置文件列表。 -如果该变量存在，则会忽略 `ENV_PROFILE`。 +如果该变量存在，则会忽略 `PROFILE_ENV`（及旧版 `ENV_PROFILE`）。 diff --git a/doc/zh/api/util.md b/doc/zh/api/util.md index 797bbd3..13a43f2 100644 --- a/doc/zh/api/util.md +++ b/doc/zh/api/util.md @@ -16,7 +16,7 @@ ## 时间工具 (`aloha.util.time`) -该模块提供用于包装函数调用（如通过 `requests` 或 `httpx` 发起外部 HTTP 请求）的超时控制工具，并在操作成功或失败（超时/异常）时触发可选的回调函数。 +该模块提供用于包装函数调用（如通过 `httpx2` 发起外部 HTTP 请求）的超时控制工具，并在操作成功或失败（超时/异常）时触发可选的回调函数。 ### 核心函数 - `run_with_timeout`: 以同步方式运行函数，并应用超时限制。 @@ -25,7 +25,7 @@ ### 使用示例 ```python from aloha.util.time import run_with_timeout -import requests +import httpx2 def success_callback(response): print("请求成功:", response.status_code) @@ -36,7 +36,7 @@ def fail_callback(exception): # 同步超时包装调用 try: run_with_timeout( - requests.get, + httpx2.get, 2.5, # 2.5 秒超时限制 "https://httpbin.org/delay/1", fn_callback_success=success_callback, diff --git a/pkg/aloha/config/hocon.py b/pkg/aloha/config/hocon.py index e3d0509..4221bb0 100644 --- a/pkg/aloha/config/hocon.py +++ b/pkg/aloha/config/hocon.py @@ -25,9 +25,7 @@ def patched_fixup_self_references(cls, config, accept_unresolved=False): if prop_path[0] == key: if isinstance(previous_item, ConfigValues) and not accept_unresolved: raise ConfigSubstitutionException( - "Property {variable} cannot be substituted. Check for cycles.".format( - variable=substitution.variable - ) + f"Property {substitution.variable} cannot be substituted. Check for cycles." ) else: value = previous_item if len(prop_path) == 1 else previous_item.get(".".join(prop_path[1:])) @@ -76,7 +74,7 @@ def load_config_from_hocon_files(config_files: list, base_dir: str): """ s = [] for config_file in config_files: - f = 'include required("%s")' % config_file + f = f'include required("{config_file}")' s.append(f) f = "\n".join(s) diff --git a/pkg/aloha/config/paths.py b/pkg/aloha/config/paths.py index 8875331..841c5d2 100644 --- a/pkg/aloha/config/paths.py +++ b/pkg/aloha/config/paths.py @@ -2,7 +2,7 @@ import sys import warnings -__all__ = ("get_resource_dir", "get_config_dir", "get_current_module_dir", "get_project_base_dir", "path_join") +__all__ = ("get_config_dir", "get_current_module_dir", "get_project_base_dir", "get_resource_dir", "path_join") def path_join(*args) -> str: @@ -60,17 +60,35 @@ def get_config_files() -> list: 1. The function will look up the `FILES_CONFIG` environment variable to get a list of file names seperated by comma, if specified; 2. In case `FILES_CONFIG` is not specified, the function will use the default config file; 3. The default config file is determined by: - (a) If environment variable `ENV_PROFILE` is defined, the entry file will be "main-{ENV_PROFILE}.conf" - (b) If environment variable `ENV_PROFILE` is not defined, the entry config file will be "main.conf". + (a) If environment variable `PROFILE_ENV` is defined, the entry file will be "main-{PROFILE_ENV}.conf" + (b) If `PROFILE_ENV` is not defined, fallback to `ENV_PROFILE` (deprecated, support will be removed in a future release); + (c) If neither is defined, the entry config file will be "main.conf". :return: list of string, which are file names of config files """ files_config = os.environ.get("FILES_CONFIG", None) if files_config is None: - env_profile = os.environ.get("ENV_PROFILE", None) - if env_profile is None: + profile_env = os.environ.get("PROFILE_ENV") + if profile_env is not None and len(profile_env.strip()) == 0: + profile_env = None + + if profile_env is None: + env_profile = os.environ.get("ENV_PROFILE") + if env_profile is not None and len(env_profile.strip()) == 0: + env_profile = None + + if env_profile is not None: + warnings.warn( + "The 'ENV_PROFILE' environment variable is deprecated and will be removed in a future release. " + "Please use 'PROFILE_ENV' instead.", + DeprecationWarning, + stacklevel=2, + ) + profile_env = env_profile + + if profile_env is None: files_config = "main.conf" else: - files_config = "main-%s.conf" % env_profile + files_config = f"main-{profile_env}.conf" files = files_config.split(",") ret = [] @@ -78,9 +96,9 @@ def get_config_files() -> list: for f in files: file = get_config_dir(f) if not os.path.exists(file): - msgs.append("Expecting config file [%s] but it does not exists!" % file) + msgs.append(f"Expecting config file [{file}] but it does not exists!") else: - print(" ---> Loading config file [%s]" % file, file=sys.stderr) + print(f" ---> Loading config file [{file}]", file=sys.stderr) ret.append(os.path.expandvars(f)) if len(ret) == 0: msgs.append("No config files set properly, EMPTY config will be used!") diff --git a/pkg/aloha/db/base.py b/pkg/aloha/db/base.py index f508d53..b3639c0 100644 --- a/pkg/aloha/db/base.py +++ b/pkg/aloha/db/base.py @@ -1,3 +1,5 @@ +from typing import ClassVar + from ..encrypt import vault from ..logger import LOG from ..settings import SETTINGS @@ -10,7 +12,7 @@ class PasswordVault: Caches vault instances for performance. """ - _dict_cache_vault = {} + _dict_cache_vault: ClassVar[dict] = {} @staticmethod def get_vault(vault_type: str | None = None, vault_config: dict | None = None, **kwargs) -> vault.BaseVault: @@ -32,7 +34,7 @@ def get_vault(vault_type: str | None = None, vault_config: dict | None = None, * encryption_method = vault_type or SETTINGS.config.get("PASSWORD_ENCRYPTION") LOG.debug("Using password vault: %s", encryption_method) # nosemgrep - cache_key = "%s:%s" % (encryption_method, str(vault_config)) + cache_key = f"{encryption_method}:{vault_config!s}" if cache_key not in PasswordVault._dict_cache_vault: if encryption_method in ("plain", "aes") or encryption_method is True: v = vault.AesVault(**(vault_config or {})) @@ -42,7 +44,7 @@ def get_vault(vault_type: str | None = None, vault_config: dict | None = None, * raise RuntimeError("Missing [CYBERARK_CONFIG] in config!") v = vault.CyberArkVault(**config_cyberark) else: - msg = "Using plain password vault as unknown value of PASSWORD_ENCRYPTION=%s in config." % encryption_method + msg = f"Using plain password vault as unknown value of PASSWORD_ENCRYPTION={encryption_method} in config." LOG.info(msg) # nosemgrep v = vault.DummyVault(**(vault_config or {})) PasswordVault._dict_cache_vault[cache_key] = v @@ -59,11 +61,11 @@ def main(): import sys config_key = sys.argv[-1] - LOG.debug("Getting pwd for deploy key [deploy.%s]" % config_key) + LOG.debug(f"Getting pwd for deploy key [deploy.{config_key}]") try: db_config = SETTINGS.config["deploy"][config_key] password_vault = PasswordVault.get_vault() p = password_vault.get_password(db_config.get("password")) - LOG.debug("Decrypted PWD: %s" % p) + LOG.debug(f"Decrypted PWD: {p}") except KeyError: - LOG.error("Please make sure config key [deploy.%s] exists!" % config_key) + LOG.error(f"Please make sure config key [deploy.{config_key}] exists!") diff --git a/pkg/aloha/db/duckdb.py b/pkg/aloha/db/duckdb.py index fdc8373..e119efb 100644 --- a/pkg/aloha/db/duckdb.py +++ b/pkg/aloha/db/duckdb.py @@ -10,7 +10,7 @@ __all__ = ("DuckOperator",) -LOG.debug("duckdb version = %s, duckdb_engine = %s " % (duckdb.__version__, duckdb_engine.__version__)) +LOG.debug(f"duckdb version = {duckdb.__version__}, duckdb_engine = {duckdb_engine.__version__} ") class DuckOperator: @@ -57,7 +57,7 @@ def __init__(self, db_config, **kwargs): LOG.debug(msg) except Exception as e: LOG.exception(e) - raise RuntimeError("Failed to connect to DuckDB") + raise RuntimeError("Failed to connect to DuckDB") from e def _prepare_database(self): """Prepare the database file and its parent directory.""" @@ -72,7 +72,7 @@ def _prepare_database(self): parent_dir.mkdir(parents=True, exist_ok=True) LOG.debug(f"Created directory: {parent_dir}") except Exception as e: - raise RuntimeError(f"Failed to create directory '{parent_dir}': {e}") + raise RuntimeError(f"Failed to create directory '{parent_dir}': {e}") from e if not path_obj.exists(): if self._config["read_only"]: @@ -81,7 +81,7 @@ def _prepare_database(self): LOG.debug(f"Database file not found, creating: {path}") duckdb.connect(path).close() except Exception as e: - raise RuntimeError(f"Failed to create database file '{path}': {e}") + raise RuntimeError(f"Failed to create database file '{path}': {e}") from e def _initialize_schema(self): """Create or select the requested schema.""" @@ -101,7 +101,7 @@ def _initialize_schema(self): self.engine.connect().execute(text(f"SET schema '{self._config['schema']}'")) except Exception as e: - raise RuntimeError(f"Failed to initialize schema: {e}") + raise RuntimeError(f"Failed to initialize schema: {e}") from e @property def connection(self): diff --git a/pkg/aloha/db/elasticsearch.py b/pkg/aloha/db/elasticsearch.py index e67b4b8..1d752f2 100644 --- a/pkg/aloha/db/elasticsearch.py +++ b/pkg/aloha/db/elasticsearch.py @@ -79,7 +79,7 @@ def build_index(self, index_name=None, index_config=None, raise_if_exist=False): res = self.es.indices.create(index=index_name or self.index_name, body=index_config or self.index_config) return res else: - msg = "Index [%s] already exits" % self.index_name + msg = f"Index [{self.index_name}] already exits" if raise_if_exist: raise RuntimeError(msg) else: diff --git a/pkg/aloha/db/kafka.py b/pkg/aloha/db/kafka.py index 0817d4f..71ce718 100644 --- a/pkg/aloha/db/kafka.py +++ b/pkg/aloha/db/kafka.py @@ -10,7 +10,7 @@ __all__ = ("KafkaOperator",) -LOG.debug("Version of confluent_kafka client = %s" % kafka.__version__) +LOG.debug(f"Version of confluent_kafka client = {kafka.__version__}") class KafkaOperator: @@ -49,19 +49,19 @@ def create_topic(self, topic: str, num_partitions=3, replication_factor=1, *args fs = a.create_topics([new_topic]) # Wait for each operation to finish. - for topic, f in fs.items(): + for topic_name, f in fs.items(): try: f.result() # The result itself is None - LOG.info("Topic {} created".format(topic)) - except Exception as e: - LOG.error("Failed to create topic {}: {}".format(topic, e)) + LOG.info(f"Topic {topic_name} created") + except Exception as e: # noqa: BLE001 + LOG.error(f"Failed to create topic {topic_name}: {e}") return False finally: a.close() return True - def producer_deliver(self, topic: str, generator: typing.Iterator[str], func_callback: callable = None, *args, **kwargs): + def producer_deliver(self, topic: str, generator: typing.Iterator[str], func_callback: callable | None = None, *args, **kwargs): """Stream messages from an iterator into a Kafka topic.""" # func_callback should be a function that takes two arguments: err and msg config_producer = {**self._config} @@ -70,9 +70,9 @@ def producer_deliver(self, topic: str, generator: typing.Iterator[str], func_cal def delivery_report(err, msg): """Called once for each message produced to indicate delivery result. Triggered by poll() or flush().""" if err is not None: - LOG.error("Kafka msg delivery failed: {}".format(err)) + LOG.error(f"Kafka msg delivery failed: {err}") else: - LOG.debug("Kafka msg delivered to {} [{}]".format(msg.topic(), msg.partition())) + LOG.debug(f"Kafka msg delivered to {msg.topic()} [{msg.partition()}]") if func_callback is None: func_callback = delivery_report @@ -108,11 +108,11 @@ def consumer_generator( code = msg.error().code() if code == kafka.KafkaError._PARTITION_EOF: pass - LOG.error("Kafka consumer: {}".format(msg.error())) + LOG.error(f"Kafka consumer: {msg.error()}") continue data = msg.value().decode("utf-8") - LOG.debug("Received message: {}".format(data)) + LOG.debug(f"Received message: {data}") yield data c.close() diff --git a/pkg/aloha/db/mongo.py b/pkg/aloha/db/mongo.py index a808e96..55af69f 100644 --- a/pkg/aloha/db/mongo.py +++ b/pkg/aloha/db/mongo.py @@ -28,12 +28,12 @@ def MongoOperator(config): collection_name = config.get("collection_name") _config = {k: v for k, v in config.items() if v is not None} - key = "%s:%s:%s" % (json.dumps(_config, sort_keys=True, ensure_ascii=False), db_name or "", collection_name or "") + key = "{}:{}:{}".format(json.dumps(_config, sort_keys=True, ensure_ascii=False), db_name or "", collection_name or "") if key not in _conn: try: _conn[key] = _MongoDBOperation(_config, db_name=db_name, collection_name=collection_name) - except Exception as e: + except Exception as e: # noqa: BLE001 LOG.exception(e) return return _conn[key] @@ -60,7 +60,7 @@ def __init__(self, config, db_name=None, collection_name=None): password_vault = PasswordVault.get_vault(config.get("vault_type"), config.get("vault_config")) _config = { - "host": "mongodb://%s" % ",".join(hosts), + "host": "mongodb://{}".format(",".join(hosts)), "port": config.get("port"), "replicaSet": replicaSet, "username": config["username"], @@ -77,13 +77,13 @@ def __init__(self, config, db_name=None, collection_name=None): self.db = self.conn[db_name] if self.collection_name is not None: self.collection = self.db[self.collection_name] - except Exception as e: + except Exception as e: # noqa: BLE001 LOG.exception(e) def set_collection(self, collection_name): """Switch the active collection after verifying it exists.""" if collection_name not in self.db.list_collection_names(): - raise Exception("Collection[%s] does not exist in [%s]" % (self.collection_name, self.db_name)) + raise RuntimeError(f"Collection[{self.collection_name}] does not exist in [{self.db_name}]") self.collection_name = collection_name self.collection = self.db[self.collection_name] return True @@ -98,7 +98,7 @@ def check_and_get_collection(self, collection_name=None, raise_if_not_exists=Tru if collection_name is not None and collection_name != self.collection_name: if self.collection_name not in self.db.list_collection_names(): if raise_if_not_exists: - raise Exception("Collection [%s] does not exist in [%s]" % (self.collection_name, self.db_name)) + raise RuntimeError(f"Collection [{self.collection_name}] does not exist in [{self.db_name}]") else: pass @@ -112,7 +112,7 @@ def insert(self, doc_or_docs, check_keys=False, collection_name=None): try: collection = self.check_and_get_collection(collection_name) return collection.insert(doc_or_docs, check_keys=check_keys) - except Exception as e: + except Exception as e: # noqa: BLE001 LOG.exception(e) def insert_many(self, docs, collection_name=None): @@ -120,7 +120,7 @@ def insert_many(self, docs, collection_name=None): try: collection = self.check_and_get_collection(collection_name) return collection.insert_many(docs) - except Exception as e: + except Exception as e: # noqa: BLE001 LOG.exception(e) def insert_one(self, doc, collection_name=None): @@ -128,7 +128,7 @@ def insert_one(self, doc, collection_name=None): try: collection = self.check_and_get_collection(collection_name) return collection.insert_one(doc) - except Exception as e: + except Exception as e: # noqa: BLE001 LOG.exception(e) def delete_many(self, field_filter, collection_name=None): @@ -136,7 +136,7 @@ def delete_many(self, field_filter, collection_name=None): try: collection = self.check_and_get_collection(collection_name) return collection.delete_many(filter=field_filter) - except Exception as e: + except Exception as e: # noqa: BLE001 LOG.exception(e) def delete_one(self, field_filter, collection_name=None): @@ -144,7 +144,7 @@ def delete_one(self, field_filter, collection_name=None): try: collection = self.check_and_get_collection(collection_name) return collection.delete_one(filter=field_filter) - except Exception as e: + except Exception as e: # noqa: BLE001 LOG.exception(e) def update_one( @@ -171,7 +171,7 @@ def update_one( session=session, ) return True - except Exception as e: + except Exception as e: # noqa: BLE001 LOG.exception(e) return False @@ -198,7 +198,7 @@ def update_many( array_filters=array_filters, session=session, ) - except Exception as e: + except Exception as e: # noqa: BLE001 LOG.exception(e) def query(self, field_filter=None, sort=None, limit=40, skip=0, collection_name=None): @@ -210,7 +210,7 @@ def query(self, field_filter=None, sort=None, limit=40, skip=0, collection_name= else: result = collection.find(field_filter or {}).skip(skip).limit(limit) return result - except Exception as e: + except Exception as e: # noqa: BLE001 LOG.exception(e) def find_many(self, field_filter=None, projection=None, collection_name=None, *args, **kwargs): @@ -219,7 +219,7 @@ def find_many(self, field_filter=None, projection=None, collection_name=None, *a collection = self.check_and_get_collection(collection_name) result = collection.find(field_filter or {}, projection, *args, **kwargs) return result - except Exception as e: + except Exception as e: # noqa: BLE001 LOG.exception(e) def find_one(self, field_filter=None, projection=None, collection_name=None, *args, **kwargs): @@ -228,7 +228,7 @@ def find_one(self, field_filter=None, projection=None, collection_name=None, *ar collection = self.check_and_get_collection(collection_name) result = collection.find_one(field_filter or {}, projection, *args, **kwargs) return result - except Exception as e: + except Exception as e: # noqa: BLE001 LOG.exception(e) def count(self, field_filter=None, collection_name=None): @@ -242,7 +242,7 @@ def count(self, field_filter=None, collection_name=None): collection = self.check_and_get_collection(collection_name) result = collection.count_documents(field_filter or {}) return result - except Exception as e: + except Exception as e: # noqa: BLE001 LOG.exception(e) def check_connected(self): diff --git a/pkg/aloha/db/mysql.py b/pkg/aloha/db/mysql.py index aff7632..8d7614c 100644 --- a/pkg/aloha/db/mysql.py +++ b/pkg/aloha/db/mysql.py @@ -9,7 +9,7 @@ __all__ = ("MySqlOperator",) -LOG.debug("Version of pymysql = %s" % pymysql.__version__) +LOG.debug(f"Version of pymysql = {pymysql.__version__}") class MySqlOperator: @@ -37,7 +37,7 @@ def __init__(self, db_config, **kwargs): LOG.debug("MySQL connected: {host}:{port}/{dbname}".format(**self._config)) except Exception as e: LOG.exception(e) - raise RuntimeError("Failed to connect to MySQL") + raise RuntimeError("Failed to connect to MySQL") from e @property def connection(self): diff --git a/pkg/aloha/db/oracle.py b/pkg/aloha/db/oracle.py index 197dd9b..049cf31 100644 --- a/pkg/aloha/db/oracle.py +++ b/pkg/aloha/db/oracle.py @@ -9,7 +9,7 @@ __all__ = ("OracledbOperator",) -LOG.debug("oracledb version = %s" % oracledb.__version__) +LOG.debug(f"oracledb version = {oracledb.__version__}") class OracledbOperator: @@ -42,10 +42,10 @@ def __init__(self, db_config, **kwargs): if "lib_dir" in db_config: # use Thick mode try: oracledb.init_oracle_client(lib_dir=db_config["lib_dir"]) - LOG.info("Oracle client initialized in THICK mode from: %s" % db_config["lib_dir"]) + LOG.info("Oracle client initialized in THICK mode from: {}".format(db_config["lib_dir"])) except Exception as e: LOG.warning(f"Warning: {e}") - raise RuntimeError(f"Failed to initialize Oracle client: {e}") + raise RuntimeError(f"Failed to initialize Oracle client: {e}") from e service_name = db_config.get("service_name") sid = db_config.get("sid") @@ -71,7 +71,7 @@ def __init__(self, db_config, **kwargs): print(msg) except Exception as e: LOG.error(e) - raise RuntimeError("Failed to connect to OracleDB") + raise RuntimeError("Failed to connect to OracleDB") from e @property def connection(self): diff --git a/pkg/aloha/db/postgres.py b/pkg/aloha/db/postgres.py index ede289e..64e7992 100644 --- a/pkg/aloha/db/postgres.py +++ b/pkg/aloha/db/postgres.py @@ -9,7 +9,7 @@ __all__ = ("PostgresOperator",) -LOG.debug("postgres: psycopg version = %s" % psycopg.__version__) +LOG.debug(f"postgres: psycopg version = {psycopg.__version__}") class PostgresOperator: @@ -42,7 +42,7 @@ def __init__(self, db_config, **kwargs): LOG.debug("PostgresSQL connected: {host}:{port}/{dbname}".format(**self._config)) except Exception as e: LOG.error(e) - raise RuntimeError("Failed to connect to PostgresSQL") + raise RuntimeError("Failed to connect to PostgresSQL") from e @property def connection(self): diff --git a/pkg/aloha/db/redis.py b/pkg/aloha/db/redis.py index 11fdb1d..4f48c6b 100644 --- a/pkg/aloha/db/redis.py +++ b/pkg/aloha/db/redis.py @@ -42,8 +42,8 @@ def _check_redis_version() -> bool: ver_cur = version.parse(redis.__version__) if ver_cur >= ver_min: valid = True - LOG.debug("Using redis version = %s" % redis.__version__) - except Exception as e: + LOG.debug(f"Using redis version = {redis.__version__}") + except Exception as e: # noqa: BLE001 LOG.error("Failed to obtain redis version!") LOG.error(str(e)) diff --git a/pkg/aloha/db/sqlite.py b/pkg/aloha/db/sqlite.py index 8e7c460..41b60ff 100644 --- a/pkg/aloha/db/sqlite.py +++ b/pkg/aloha/db/sqlite.py @@ -19,7 +19,7 @@ def __init__(self, db_config, **kwargs): self._connection_pattern = "sqlite://{dbname}" dbname = db_config.get("dbname", "") if len(dbname) > 0: - dbname = "/%s" % dbname + dbname = f"/{dbname}" self._config = {"dbname": dbname} if "password" in db_config: @@ -27,20 +27,20 @@ def __init__(self, db_config, **kwargs): import sqlcipher3 except ImportError: raise RuntimeError("Python package required for encrypted sqlite3: sqlcipher3-binary") - LOG.debug("Version of sqlcipher3 = %s" % sqlcipher3.sqlite_version) + LOG.debug(f"Version of sqlcipher3 = {sqlcipher3.sqlite_version}") password_vault = PasswordVault.get_vault(db_config.get("vault_type"), db_config.get("vault_config")) password = password_vault.get_password(db_config.get("password", None)) self._config["password"] = password self._connection_pattern = "sqlite+pysqlcipher://:{password}@/{dbname}" else: - LOG.debug("Version of sqlite = %s" % sqlite3.sqlite_version) + LOG.debug(f"Version of sqlite = {sqlite3.sqlite_version}") try: self.db = create_engine(self._connection_pattern.format(**self._config), **kwargs) - LOG.debug("Sqlite connected: %s" % self.connection_str) + LOG.debug(f"Sqlite connected: {self.connection_str}") except Exception as e: LOG.exception(e) - raise RuntimeError("Failed to connect to sqlite") + raise RuntimeError("Failed to connect to sqlite") from e @property def connection(self): diff --git a/pkg/aloha/encrypt/aes.py b/pkg/aloha/encrypt/aes.py index 36c0924..e39e454 100644 --- a/pkg/aloha/encrypt/aes.py +++ b/pkg/aloha/encrypt/aes.py @@ -2,7 +2,7 @@ import base64 import binascii -from typing import Callable, Optional, Union +from collections.abc import Callable from Crypto.Cipher import AES from Crypto.Random import get_random_bytes @@ -22,7 +22,7 @@ def _generate_key(key_size: int, method="const") -> bytes: return b"0" * key_size # b'b6046801716aec00' elif method == "random": return get_random_bytes(key_size) - raise ValueError("Invalid AES key generate method: [%s]" % method) + raise ValueError(f"Invalid AES key generate method: [{method}]") class AesEncryptor: @@ -30,7 +30,7 @@ class AesEncryptor: supported_cipher_methods = _AES_CIPHER_METHODS - def __init__(self, key: Union[str, bytes] = None, key_size: int = 16, cipher_name: str = "AES/ECB/PKCS5Padding"): + def __init__(self, key: str | bytes | None = None, key_size: int = 16, cipher_name: str = "AES/ECB/PKCS5Padding"): """Initialize the AES key and cipher settings.""" _key = key if key is None: @@ -43,13 +43,13 @@ def __init__(self, key: Union[str, bytes] = None, key_size: int = 16, cipher_nam 24, 32, ): - raise ValueError("Invalid key size/length [%s] for AesEncryptor!" % len(_key)) + raise ValueError(f"Invalid key size/length [{len(_key)}] for AesEncryptor!") self.key_aes, self.block_size = _key, AES.block_size # https://pycryptodome.readthedocs.io/en/latest/src/util/util.html self.cipher_name = cipher_name - def encrypt(self, text: str, output_format="hex", func_pad: Optional[Callable] = None) -> Union[str, bytes]: + def encrypt(self, text: str, output_format="hex", func_pad: Callable | None = None) -> str | bytes: """Encrypt a UTF-8 string and return hex, base64, or raw bytes.""" dict_params, pad_style = _AES_CIPHER_METHODS.get(self.cipher_name) if not callable(func_pad): @@ -72,12 +72,12 @@ def _func_pad(x): elif output_format in ("bytes", "bin"): crypt = bytes_crypt else: - raise ValueError("Unknown output_type [%s]" % output_format) + raise ValueError(f"Unknown output_type [{output_format}]") return crypt def decrypt( - self, text: Union[str, bytes], input_format: str = "hex", func_unpad: Optional[Callable] = None - ) -> Union[str, bytes]: + self, text: str | bytes, input_format: str = "hex", func_unpad: Callable | None = None + ) -> str | bytes: """Decrypt ciphertext produced by :meth:`encrypt`.""" text += (len(text) % 4) * "=" if input_format == "hex": @@ -87,7 +87,7 @@ def decrypt( elif input_format in ("bytes", "bin"): crypt = text else: - raise ValueError("Unknown output_type [%s]" % input_format) + raise ValueError(f"Unknown output_type [{input_format}]") dict_params, pad_style = _AES_CIPHER_METHODS.get(self.cipher_name) cipher = AES.new(key=self.key_aes, **dict_params) data = cipher.decrypt(crypt) diff --git a/pkg/aloha/encrypt/jwt.py b/pkg/aloha/encrypt/jwt.py index d732cde..0ce5af1 100644 --- a/pkg/aloha/encrypt/jwt.py +++ b/pkg/aloha/encrypt/jwt.py @@ -4,7 +4,7 @@ from ..logger import LOG -LOG.debug("Using pyjwt == %s" % str(jwt.__version__)) +LOG.debug(f"Using pyjwt == {jwt.__version__!s}") def encode(secret_key: str, payload: dict, headers: dict | None = None, **kwargs): diff --git a/pkg/aloha/encrypt/rsa.py b/pkg/aloha/encrypt/rsa.py index 8c95953..82e3583 100644 --- a/pkg/aloha/encrypt/rsa.py +++ b/pkg/aloha/encrypt/rsa.py @@ -1,8 +1,7 @@ """RSA encrypt/decrypt and signing helpers.""" import base64 -from functools import lru_cache -from typing import Optional, Tuple, Union +from typing import ClassVar from Crypto.Cipher import PKCS1_OAEP, PKCS1_v1_5 from Crypto.Hash import SHA1, SHA256 @@ -11,7 +10,7 @@ __all__ = ("RsaEncryptor",) -t_cipher_module = Union[PKCS1_v1_5.PKCS115_Cipher, PKCS1_OAEP.PKCS1OAEP_Cipher] +t_cipher_module = PKCS1_v1_5.PKCS115_Cipher | PKCS1_OAEP.PKCS1OAEP_Cipher _RSA_CIPHER_METHODS = { # FULL_CIPHER_NAME: (module, dict_params) "RSA/ECB/PKCS1Padding": (PKCS1_v1_5, {"randfunc": None}), @@ -23,8 +22,8 @@ class RsaEncryptor: """Encrypt, decrypt, and convert RSA keys and payloads.""" - _dict_cache_cipher = {} - _dict_cache_decipher = {} + _dict_cache_cipher: ClassVar[dict] = {} + _dict_cache_decipher: ClassVar[dict] = {} supported_cipher_methods = _RSA_CIPHER_METHODS # ref: https://cryptobook.nakov.com/asymmetric-key-ciphers/rsa-encrypt-decrypt-examples @@ -37,20 +36,19 @@ def __init__( self.cipher_name = cipher_name @staticmethod - def _get_cipher_module(full_cipher_name: str | None = None) -> Optional[Tuple]: + def _get_cipher_module(full_cipher_name: str | None = None) -> tuple | None: try: return _RSA_CIPHER_METHODS[full_cipher_name] except KeyError: - raise ValueError("Unsupported full cipher name, supported ones: %s." % ",".join(sorted(_RSA_CIPHER_METHODS))) + raise ValueError("Unsupported full cipher name, supported ones: {}.".format(",".join(sorted(_RSA_CIPHER_METHODS)))) @staticmethod - def generate_key_pair(size: int = 1024) -> Tuple[str, str]: + def generate_key_pair(size: int = 1024) -> tuple[str, str]: """Generate a PEM-encoded RSA key pair.""" key_pair = RSA.generate(size) key_private, key_public = key_pair.exportKey(), key_pair.publickey().exportKey() return key_private.decode("ascii"), key_public.decode("ascii") - @lru_cache def get_cipher(self, key_public: str | None = None, cipher_name="RSA/ECB/PKCS1Padding") -> t_cipher_module: """Return a cached public-key cipher instance.""" if key_public is None: @@ -66,7 +64,6 @@ def get_cipher(self, key_public: str | None = None, cipher_name="RSA/ECB/PKCS1Pa # debug: print('->PUB', cache_key, len(RsaEncryptor._dict_cache_cipher)) return RsaEncryptor._dict_cache_cipher[cache_key] - @lru_cache def get_decipher(self, key_private: str | None = None, cipher_name="RSA/ECB/PKCS1Padding") -> t_cipher_module: """Return a cached private-key decipher instance.""" if key_private is None: @@ -85,7 +82,7 @@ def get_decipher(self, key_private: str | None = None, cipher_name="RSA/ECB/PKCS @staticmethod def load_keys_from_binary( key_private: bytes | None = None, key_public: bytes | None = None - ) -> Tuple[Optional[RSA.RsaKey], Optional[RSA.RsaKey]]: + ) -> tuple[RSA.RsaKey | None, RSA.RsaKey | None]: """Load RSA keys from PEM/DER bytes.""" _key_private, _key_public = None, None @@ -93,20 +90,20 @@ def load_keys_from_binary( try: _key_private = RSA.import_key(key_private) except ValueError: - raise ValueError("RSA pri key format error: [%s]" % key_private) + raise ValueError(f"RSA pri key format error: [{key_private}]") if key_public is not None: try: _key_public = RSA.import_key(key_public) except ValueError: - raise ValueError("RSA pub key format error: [%s]" % key_public) + raise ValueError(f"RSA pub key format error: [{key_public}]") return _key_private, _key_public @staticmethod def load_keys_from_string( key_private: str | None = None, key_public: str | None = None - ) -> Tuple[Optional[RSA.RsaKey], Optional[RSA.RsaKey]]: + ) -> tuple[RSA.RsaKey | None, RSA.RsaKey | None]: """Load RSA keys from PEM-like strings.""" _key_private, _key_public = None, None @@ -118,7 +115,7 @@ def load_keys_from_string( try: _key_private = RSA.import_key(key_pri) except ValueError: - raise ValueError("RSA private key format error: [%s]" % key_pri) + raise ValueError(f"RSA private key format error: [{key_pri}]") if key_public is not None: if not key_public.startswith("-----"): @@ -128,12 +125,12 @@ def load_keys_from_string( try: _key_public = RSA.import_key(key_pub) except ValueError: - raise ValueError("RSA public key format error: [%s]" % key_pub) + raise ValueError(f"RSA public key format error: [{key_pub}]") return _key_private, _key_public def encrypt_with_public_key( - self, message: Union[str, bytes], key_public: str | None = None, cipher_name: str = None + self, message: str | bytes, key_public: str | None = None, cipher_name: str | None = None ) -> bytes: """Encrypt a message with a public key.""" data = message if isinstance(message, bytes) else message.encode("UTF-8") @@ -141,7 +138,7 @@ def encrypt_with_public_key( return cipher.encrypt(data) def decrypt_with_private_key( - self, ciphertext: Union[str, bytes], key_private: str | None = None, cipher_name: str | None = None, **kwargs + self, ciphertext: str | bytes, key_private: str | None = None, cipher_name: str | None = None, **kwargs ) -> bytes: """Decrypt ciphertext with a private key.""" data = ciphertext if isinstance(ciphertext, bytes) else ciphertext.encode("ascii") @@ -176,7 +173,5 @@ def main(): y_dec = rsa_dec.decrypt_with_private_key(y_bin, key_private=key_pri) y_txt = y_dec.decode("UTF-8") - msg = "[test {i_case} success = {status}] {src} -> {enc}".format( - i_case=i, status=(y_txt == str_src), src=y_txt, enc=x_txt - ) + msg = f"[test {i} success = {y_txt == str_src}] {y_txt} -> {x_txt}" print(msg) diff --git a/pkg/aloha/encrypt/vault/__init__.py b/pkg/aloha/encrypt/vault/__init__.py index 2ccd294..12ba0a4 100644 --- a/pkg/aloha/encrypt/vault/__init__.py +++ b/pkg/aloha/encrypt/vault/__init__.py @@ -2,4 +2,4 @@ from .cyberark import CyberArkVault from .plain import AesVault -__all__ = ("BaseVault", "DummyVault", "AesVault", "CyberArkVault") +__all__ = ("AesVault", "BaseVault", "CyberArkVault", "DummyVault") diff --git a/pkg/aloha/encrypt/vault/cyberark.py b/pkg/aloha/encrypt/vault/cyberark.py index e6c7954..3396fdd 100644 --- a/pkg/aloha/encrypt/vault/cyberark.py +++ b/pkg/aloha/encrypt/vault/cyberark.py @@ -2,25 +2,21 @@ import hashlib from binascii import a2b_hex +from typing import ClassVar from urllib.parse import quote_plus as urlquote -import requests +import httpx2 from Crypto.Cipher import AES -from requests.packages.urllib3.exceptions import InsecureRequestWarning from ...encrypt.aes import AesEncryptor from ...logger import LOG from .base import BaseVault -requests.packages.urllib3.disable_warnings(InsecureRequestWarning) -if hasattr(requests.packages.urllib3.util.ssl_, "DEFAULT_CIPHERS"): - requests.packages.urllib3.util.ssl_.DEFAULT_CIPHERS += ":HIGHT:!DH:!aNULL" - class CyberArkVault(BaseVault, AesEncryptor): """Fetch and decrypt passwords from a CyberArk-compatible endpoint.""" - _cached: dict = {} + _cached: ClassVar[dict] = {} def __init__(self, url: str, app_id: str, key: str | None = None, safe: str = "AIM_ELIS_LAS", folder: str = "root"): """Initialize the vault with the CyberArk endpoint and credentials.""" @@ -63,7 +59,7 @@ def get_cyberark_password(self, object: str | None = None, **kwargs): while retry: try: LOG.debug("POST CyberArk: %s with data: %s", self.url, data) - resp = requests.post( + resp = httpx2.post( self.url, json=data, headers={"Content-Type": "application/json"}, @@ -78,16 +74,14 @@ def get_cyberark_password(self, object: str | None = None, **kwargs): except Exception as e: retry -= 1 if retry == 0: - raise e + raise else: - LOG.error("CyberArk request error: {}".format(e)) + LOG.error(f"CyberArk request error: {e}") return None def get_password(self, object=None, **kwargs): """Return a cached CyberArk password, optionally URL-encoded.""" - key_for_cache = "{app_id};{safe};{folder};{key};{object}".format( - app_id=self.app_id, safe=self.safe, folder=self.folder, key=self.key, object=object - ) + key_for_cache = f"{self.app_id};{self.safe};{self.folder};{self.key};{object}" if key_for_cache not in self._cached: kwargs.update(object if isinstance(object, dict) else {"object": object}) url_quote = kwargs.get("url_encode", True) @@ -97,20 +91,20 @@ def get_password(self, object=None, **kwargs): pwd = urlquote(pwd) self._cached[key_for_cache] = pwd else: - LOG.debug("Using cached CyberArk key: %s" % key_for_cache) + LOG.debug(f"Using cached CyberArk key: {key_for_cache}") return self._cached[key_for_cache] def main(): """Small self-test scaffold for the CyberArk vault.""" - cfg_cyberark = dict( - url="https://localhost/pidms/rest/pwd/getPassword", # to fill properly - app_id="", - safe="", - folder="root", - key="", - ) + cfg_cyberark = { + "url": "https://localhost/pidms/rest/pwd/getPassword", # to fill properly + "app_id": "", + "safe": "", + "folder": "root", + "key": "", + } # from ...settings import SETTINGS # cfg_cyberark = SETTINGS.config['CYBERARK_CONFIG'] vault = CyberArkVault(**cfg_cyberark) diff --git a/pkg/aloha/logger/__init__.py b/pkg/aloha/logger/__init__.py index 1459143..ade112a 100644 --- a/pkg/aloha/logger/__init__.py +++ b/pkg/aloha/logger/__init__.py @@ -4,4 +4,4 @@ LOG = get_logger( level=SETTINGS.config.get("deploy", {}).get("log_level", 10), # 10 = logging.DEBUG ) -__all__ = ("LOG", "get_logger", "getLogger") +__all__ = ("LOG", "getLogger", "get_logger") diff --git a/pkg/aloha/logger/handler.py b/pkg/aloha/logger/handler.py index 447cfcb..fd59779 100644 --- a/pkg/aloha/logger/handler.py +++ b/pkg/aloha/logger/handler.py @@ -18,9 +18,7 @@ def __init__(self, filename: str, encoding="utf8", delay=False, utc=False, **kwa BaseRotatingHandler.__init__(self, filename, "a", encoding, delay) def shouldRollover(self, record): - if self.currentFileName != self._compute_fn(): - return True - return False + return self.currentFileName != self._compute_fn() def doRollover(self): if self.stream: diff --git a/pkg/aloha/logger/logger.py b/pkg/aloha/logger/logger.py index 5bb0160..91af830 100644 --- a/pkg/aloha/logger/logger.py +++ b/pkg/aloha/logger/logger.py @@ -41,9 +41,9 @@ def setup_logger( if logger_name is not None and len(logger_name) > 0: logger_name = logger_name.strip().replace(" ", "_") - path_file = [module, logger_name, socket.gethostname(), "p%s" % os.getpid()] # module, logger_name, hostname, pid + path_file = [module, logger_name, socket.gethostname(), f"p{os.getpid()}"] # module, logger_name, hostname, pid path_file = "_".join(str(i) for i in path_file if i is not None and len(str(i)) > 0) - path_file = pjoin(folder, "%s.log" % path_file) + path_file = pjoin(folder, f"{path_file}.log") file_handler = MultiProcessSafeDailyRotatingFileHandler(path_file) file_handler.setFormatter(formatter) diff --git a/pkg/aloha/script/base.py b/pkg/aloha/script/base.py index 1a05a92..14b3380 100644 --- a/pkg/aloha/script/base.py +++ b/pkg/aloha/script/base.py @@ -12,19 +12,19 @@ def main(): args, _ = parser.parse_known_args() cmd = args.cmd - module = "%s.%s" % (__package__, cmd) + module = f"{__package__}.{cmd}" try: module = importlib.import_module(module) except ImportError as e: - print("Invalid sub-command: %s\n\tFailed to import: %s" % (cmd, module)) + print(f"Invalid sub-command: {cmd}\n\tFailed to import: {module}") print(str(e)) - exit(-1) + sys.exit(-1) sys.argv.pop(0) - print("aloha command options: %s" % "".join(sys.argv)) - func_main = getattr(module, "main") + print("aloha command options: {}".format("".join(sys.argv))) + func_main = module.main - exit(func_main()) + sys.exit(func_main()) if __name__ == "__main__": diff --git a/pkg/aloha/script/compile.py b/pkg/aloha/script/compile.py old mode 100644 new mode 100755 index 7d4f0cf..aa9e133 --- a/pkg/aloha/script/compile.py +++ b/pkg/aloha/script/compile.py @@ -30,15 +30,15 @@ def _expand(patterns: list | None = None): def _delete(file_path: str, ignore_errors=True): - print("Removing file/folder: %s" % file_path) + print(f"Removing file/folder: {file_path}") try: if os.path.isfile(file_path) or os.path.islink(file_path): os.unlink(file_path) elif os.path.isdir(file_path): shutil.rmtree(file_path) - except Exception as e: + except Exception: if not ignore_errors: - raise e + raise def build( @@ -101,7 +101,7 @@ def build( # c code -> dynamic library file path_build_tmp = os.path.join(path_build, ".tmp") script_args = ["build_ext", "-b", path_build, "-t", path_build_tmp, "-j", n_parallel] - print("Build args: %s" % " ".join(str(s) for s in script_args)) + print("Build args: {}".format(" ".join(str(s) for s in script_args))) setup(ext_modules=cythonized, script_args=script_args) # clean up @@ -127,7 +127,7 @@ def package( path_dist = os.path.abspath(dist) os.makedirs(path_dist, exist_ok=True) if len(glob.glob(path_dist + "/*")) > 0: - raise ValueError("Dist folder [%s] MUST be an empty directory or an non-existing folder!" % path_dist) + raise ValueError(f"Dist folder [{path_dist}] MUST be an empty directory or an non-existing folder!") folder_name = os.getcwd().split(os.sep)[-1] folder_temp = os.path.join("/tmp/build/", folder_name) @@ -143,7 +143,7 @@ def package( ) [shutil.move(f, os.path.join(path_dist, f.split(os.sep)[-1])) for f in glob.glob(folder_temp + "/*")] t = time.time() - t - print("\n\nTime consumed to build code: %.2f seconds." % t) + print(f"\n\nTime consumed to build code: {t:.2f} seconds.") print("Successfully finished building package to: ", path_dist) @@ -159,7 +159,7 @@ def main(): args = p.parse_args() args = vars(args) for k, v in args.items(): - print("%s = %s" % (k, v)) + print(f"{k} = {v}") package(**args) diff --git a/pkg/aloha/script/info.py b/pkg/aloha/script/info.py index af053aa..e0d64e7 100644 --- a/pkg/aloha/script/info.py +++ b/pkg/aloha/script/info.py @@ -2,4 +2,4 @@ def main(**kwargs): - print('Aloha! version: %s' % __version__) + print(f'Aloha! version: {__version__}') diff --git a/pkg/aloha/script/start.py b/pkg/aloha/script/start.py index aefd0b6..45799ba 100644 --- a/pkg/aloha/script/start.py +++ b/pkg/aloha/script/start.py @@ -4,7 +4,7 @@ def main(): - print("\n".join(sorted("%s=%s" % (k, v) for k, v in os.environ.items()))) + print("\n".join(sorted(f"{k}={v}" for k, v in os.environ.items()))) usage = """ Usage: `python main.py app_common.main` ; or set environment variable `ENTRYPOINT` @@ -18,19 +18,19 @@ def main(): if module_name is None: print(usage) - exit(-1) + sys.exit(-1) try: m = importlib.import_module(module_name) except ImportError: - raise ValueError("Invalid entrypoint: %s" % module_name) + raise ValueError(f"Invalid entrypoint: {module_name}") - f_main = getattr(m, "main") + f_main = m.main if f_main is None: print("Given module does not provides a `main()` function!") else: - print("Starting module: %s" % module_name) + print(f"Starting module: {module_name}") ret = f_main() if ret: print(ret) diff --git a/pkg/aloha/service/api/v0.py b/pkg/aloha/service/api/v0.py index fcea0a9..8ddc707 100644 --- a/pkg/aloha/service/api/v0.py +++ b/pkg/aloha/service/api/v0.py @@ -5,8 +5,10 @@ serialized as a JSON object with a `code` and `message` field. """ +import json import logging from abc import ABC +from typing import ClassVar from fastapi import Request from fastapi.responses import JSONResponse @@ -14,7 +16,7 @@ from ..http import AbstractApiClient from ..http.base_api_handler import AbstractApiHandler as BaseHandler -__all__ = ("APIHandler", "APICaller", "create_v0_router") +__all__ = ("APICaller", "APIHandler", "create_v0_router") class APIHandler(BaseHandler, ABC): @@ -24,7 +26,7 @@ class APIHandler(BaseHandler, ABC): and returns a Python object that can be JSON-serialized. """ - MAP_ERROR_INFO = {"BAD_REQUEST": {"code": "5101", "message": ["Bad request: fail to parse body as JSON object!"]}} + MAP_ERROR_INFO: ClassVar[dict] = {"BAD_REQUEST": {"code": "5101", "message": ["Bad request: fail to parse body as JSON object!"]}} async def post(self, *args, **kwargs): """Parse the request body, call :meth:`response`, and return JSON.""" @@ -33,13 +35,13 @@ async def post(self, *args, **kwargs): if req_body is not None: kwargs.update(req_body) - resp = dict(code=5200, message=["success"]) + resp = {"code": 5200, "message": ["success"]} try: result = self.response(*args, **kwargs) resp["data"] = result except Exception as e: if self.LOG.level == logging.DEBUG: - self.LOG.error(e, exc_info=True) + self.LOG.exception("Error processing POST request") return self.finish({"code": 5201, "message": [repr(e)]}) return self.finish(resp) @@ -47,13 +49,13 @@ async def post(self, *args, **kwargs): async def get(self, *args, **kwargs): """Handle GET request (useful for some v0 endpoints).""" kwargs.update(self.request_param) - resp = dict(code=5200, message=["success"]) + resp = {"code": 5200, "message": ["success"]} try: result = self.response(*args, **kwargs) resp["data"] = result except Exception as e: if self.LOG.level == logging.DEBUG: - self.LOG.error(e, exc_info=True) + self.LOG.exception("Error processing GET request") return self.finish({"code": 5201, "message": [repr(e)]}) return self.finish(resp) @@ -75,17 +77,17 @@ async def handle_post(request: Request, **kwargs): # Get body for POST try: body = await request.json() - except Exception: + except (json.JSONDecodeError, ValueError): body = {} kwargs.update(body) - resp = dict(code=5200, message=["success"]) + resp = {"code": 5200, "message": ["success"]} try: result = handler.response(**kwargs) resp["data"] = result except Exception as e: if handler.LOG.level == logging.DEBUG: - handler.LOG.error(e, exc_info=True) + handler.LOG.exception("Error in handle_post") return JSONResponse({"code": 5201, "message": [repr(e)]}, status_code=500) return JSONResponse(resp) @@ -96,13 +98,13 @@ async def handle_get(request: Request, **kwargs): # Get query params for GET kwargs.update(dict(request.query_params)) - resp = dict(code=5200, message=["success"]) + resp = {"code": 5200, "message": ["success"]} try: result = handler.response(**kwargs) resp["data"] = result except Exception as e: if handler.LOG.level == logging.DEBUG: - handler.LOG.error(e, exc_info=True) + handler.LOG.exception("Error in handle_get") return JSONResponse({"code": 5201, "message": [repr(e)]}, status_code=500) return JSONResponse(resp) diff --git a/pkg/aloha/service/api/v1.py b/pkg/aloha/service/api/v1.py index 6a5815c..a4584ba 100644 --- a/pkg/aloha/service/api/v1.py +++ b/pkg/aloha/service/api/v1.py @@ -8,6 +8,7 @@ import logging import uuid from abc import ABC +from typing import ClassVar from fastapi import Request from fastapi.responses import JSONResponse @@ -17,7 +18,7 @@ from ..http import AbstractApiClient from ..http.base_api_handler import AbstractApiHandler as BaseHandler -__all__ = ("APIHandler", "APICaller", "sign_data", "sign_check", "create_v1_router") +__all__ = ("APICaller", "APIHandler", "create_v1_router", "sign_check", "sign_data") APP_ID_KEYS = SETTINGS.config.get("APP_ID_KEYS", {}) APP_OPTIONS = SETTINGS.config.get("APP_OPTIONS", {}) @@ -28,7 +29,7 @@ class APIHandler(BaseHandler, ABC): """Signed API handler for v1 endpoints.""" - MAP_ERROR_INFO = { + MAP_ERROR_INFO: ClassVar[dict] = { "BAD_REQUEST": {"code": "5101", "message": ["Bad request: fail to parse body as JSON object!"]}, "MISSING_ARGS": {"code": "5102", "message": ["Required argument field(s) missing..."]}, "SIGN_CHECK_FAIL": {"code": "5104", "message": ["Invalid sign, sign check failed!"]}, @@ -50,14 +51,14 @@ async def post(self): if not is_valid_req: return self.finish(self.MAP_ERROR_INFO["SIGN_CHECK_FAIL"]) - resp = dict(code=5200, message=["success"]) + resp = {"code": 5200, "message": ["success"]} try: result = self.response(**data) resp["data"] = result resp["salt_uuid"] = salt_uuid except Exception as e: if self.LOG.level == logging.DEBUG: - self.LOG.error(e, exc_info=True) + self.LOG.exception("Error processing v1 request") return self.finish({"code": 5201, "message": [repr(e)]}) return self.finish(resp) @@ -76,7 +77,7 @@ def create_v1_router(handler_class): async def handle_post(request: Request, **kwargs): try: body = await request.json() - except Exception: + except (json.JSONDecodeError, ValueError): return JSONResponse( {"code": "5101", "message": ["Bad request: fail to parse body as JSON object!"]}, status_code=400 ) @@ -96,14 +97,14 @@ async def handle_post(request: Request, **kwargs): handler = handler_class() handler._request = request - resp = dict(code=5200, message=["success"]) + resp = {"code": 5200, "message": ["success"]} try: result = handler.response(**data) resp["data"] = result resp["salt_uuid"] = salt_uuid except Exception as e: if handler.LOG.level == logging.DEBUG: - handler.LOG.error(e, exc_info=True) + handler.LOG.exception("Error in handle_post") return JSONResponse({"code": 5201, "message": [repr(e)]}, status_code=500) return JSONResponse(resp) @@ -127,7 +128,7 @@ def wrap_request_data( ): """Wrap the payload with signature fields expected by v1 handlers.""" if app_id is None: - app_id = list(self.APP_ID_KEYS.keys())[0] + app_id = next(iter(self.APP_ID_KEYS.keys())) salt_uuid = salt_uuid or str(uuid.uuid1()) sign = sign or sign_data( salt_uuid=salt_uuid, @@ -139,7 +140,7 @@ def wrap_request_data( return {"salt_uuid": salt_uuid, "app_id": app_id, "sign": sign, "data": data} -def sign_data(salt_uuid: str, app_id: str, app_key: str, data, sign_method: str = None): +def sign_data(salt_uuid: str, app_id: str, app_key: str, data, sign_method: str | None = None): """Generate the v1 signature for a payload. The signature is based on `app_id + salt_uuid + data + app_key`. @@ -149,17 +150,17 @@ def sign_data(salt_uuid: str, app_id: str, app_key: str, data, sign_method: str func_sign_check = func_sign_check_default if sign_method is None else FUNC_SIGN_CHECK.get(sign_method) if func_sign_check is None: - raise ValueError("Invalid `sign_method`: %s" % sign_method) + raise ValueError(f"Invalid `sign_method`: {sign_method}") sign = func_sign_check(public_key) return sign -def sign_check(salt_uuid: str, app_id: str, sign: str, data, sign_method: str = None, date_time=None): +def sign_check(salt_uuid: str, app_id: str, sign: str, data, sign_method: str | None = None, date_time=None): """Validate a v1 request signature.""" func_sign_check = func_sign_check_default if sign_method is None else FUNC_SIGN_CHECK.get(sign_method) if func_sign_check is None: - raise ValueError("Invalid `sign_method`: %s" % sign_method) + raise ValueError(f"Invalid `sign_method`: {sign_method}") app_key = APP_ID_KEYS.get(app_id) if app_key is None: diff --git a/pkg/aloha/service/api/v2.py b/pkg/aloha/service/api/v2.py index f878afa..deaad96 100644 --- a/pkg/aloha/service/api/v2.py +++ b/pkg/aloha/service/api/v2.py @@ -8,8 +8,8 @@ import json import logging from abc import ABC -from datetime import datetime, timedelta -from typing import Any, Dict, Optional +from datetime import datetime, timedelta, timezone +from typing import Annotated, Any from fastapi import Depends, HTTPException, Request, Response, status from fastapi.responses import JSONResponse @@ -20,13 +20,13 @@ from ..http import AbstractApiClient from ..http.base_api_handler import AbstractApiHandler as BaseHandler -__all__ = ("APIHandler", "APICaller", "create_v2_router", "verify_v2_token") +__all__ = ("APICaller", "APIHandler", "create_v2_router", "verify_v2_token") class APIHandler(BaseHandler, ABC): """Token-authenticated API handler for v2 endpoints.""" - async def prepare(self) -> Optional[Response]: + async def prepare(self) -> Response | None: """Validate the access token before handling the request.""" access_token = self._request.headers.get("Access-Token") if access_token is None: @@ -36,7 +36,7 @@ async def prepare(self) -> Optional[Response]: options = {"verify_exp": False} access_token = jwt.decode(secret_key, access_token, options=options) if not isinstance(access_token, dict): - msg = "Invalid Access-Token found in request for [%s]: %s" % (str(self._request.url), access_token) + msg = f"Invalid Access-Token found in request for [{self._request.url!s}]: {access_token}" self.LOG.error(msg) return self.finish({"msg": msg}) return None @@ -48,13 +48,13 @@ async def post(self, *args, **kwargs): try: if self.LOG.level == logging.DEBUG: s_kwargs = json.dumps(kwargs, ensure_ascii=False) - self.LOG.debug("POST Request [%s]: %s" % (self.request_id, s_kwargs[:1000])) + self.LOG.debug(f"POST Request [{self.request_id}]: {s_kwargs[:1000]}") self.api_args, self.api_kwargs = args or (), kwargs or {} resp = self.response(*self.api_args, **self.api_kwargs) except Exception as e: - self.LOG.info("POST Request [%s]: %s" % (self.request_id, self._request._body)) + self.LOG.info(f"POST Request [{self.request_id}]: {self._request._body}") msgs = ["An internal error has occurred!", str(e)] - self.LOG.error(e, exc_info=True) + self.LOG.exception("Error processing POST request") return self.finish({"status": "error", "message": msgs}) return self.finish(resp) @@ -64,19 +64,19 @@ async def get(self, *args, **kwargs): query_arguments = self.request_param kwargs.update(query_arguments) try: - self.LOG.debug("GET Request [%s]: %s" % (self.request_id, kwargs)) + self.LOG.debug(f"GET Request [{self.request_id}]: {kwargs}") self.api_args, self.api_kwargs = args or (), kwargs or {} resp = self.response(*self.api_args, **self.api_kwargs) except Exception as e: - self.LOG.info("GET Request [%s]: %s" % (self.request_id, kwargs)) + self.LOG.info(f"GET Request [{self.request_id}]: {kwargs}") msgs = ["An internal error has occurred!", str(e)] - self.LOG.error(e, exc_info=True) + self.LOG.exception("Error processing GET request") return self.finish({"status": "error", "message": msgs}) return self.finish(resp) -def verify_v2_token(request: Request) -> Optional[Dict[str, Any]]: +def verify_v2_token(request: Request) -> dict[str, Any] | None: """Dependency to verify v2 access token. Returns the decoded token payload if valid, otherwise raises HTTPException. @@ -96,9 +96,11 @@ def verify_v2_token(request: Request) -> Optional[Dict[str, Any]]: if not isinstance(payload, dict): raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid Access-Token!") return payload + except HTTPException: + raise except Exception as e: - LOG.error(str(e), exc_info=True) - raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid Access-Token!") + LOG.exception("Error validating v2 token") + raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid Access-Token!") from e def create_v2_router(handler_class): @@ -111,39 +113,39 @@ def create_v2_router(handler_class): Tuple of (handle_post, handle_get) functions for the routes """ - async def handle_post(request: Request, token_payload: Dict = Depends(verify_v2_token)): + async def handle_post(request: Request, token_payload: Annotated[dict, Depends(verify_v2_token)]): handler = handler_class() handler._request = request try: body = await request.json() - except Exception: + except (json.JSONDecodeError, ValueError): body = {} kwargs = body try: if handler.LOG.level == logging.DEBUG: s_kwargs = json.dumps(kwargs, ensure_ascii=False) - handler.LOG.debug("POST Request [%s]: %s" % (handler.request_id, s_kwargs[:1000])) + handler.LOG.debug(f"POST Request [{handler.request_id}]: {s_kwargs[:1000]}") resp = handler.response(**kwargs) except Exception as e: - handler.LOG.error(e, exc_info=True) + handler.LOG.exception("Error in handle_post") msgs = ["An internal error has occurred.", str(e)] return JSONResponse({"status": "error", "message": msgs}, status_code=500) return handler.finish(resp) - async def handle_get(request: Request, token_payload: Dict = Depends(verify_v2_token)): + async def handle_get(request: Request, token_payload: Annotated[dict, Depends(verify_v2_token)]): handler = handler_class() handler._request = request kwargs = dict(request.query_params) try: - handler.LOG.debug("GET Request [%s]: %s" % (handler.request_id, kwargs)) + handler.LOG.debug(f"GET Request [{handler.request_id}]: {kwargs}") resp = handler.response(**kwargs) except Exception as e: - handler.LOG.error(e, exc_info=True) + handler.LOG.exception("Error in handle_get") msgs = ["An internal error has occurred.", repr(e)] return JSONResponse({"status": "error", "message": msgs}, status_code=500) @@ -163,12 +165,12 @@ def wrap_request_data(self, data: dict) -> dict: assert isinstance(data, dict), "Data object must be a dict!" return data - def get_headers(self, app_id: str = None, app_key: str = None) -> dict: + def get_headers(self, app_id: str | None = None, app_key: str | None = None) -> dict: """Build the HTTP headers expected by v2 handlers.""" if app_id is None: - app_id = list(self.APP_ID_KEYS.keys())[0] + app_id = next(iter(self.APP_ID_KEYS.keys())) - expire_time = datetime.now() + timedelta(days=1) + expire_time = datetime.now(tz=timezone.utc) + timedelta(days=1) access_token = jwt.encode(secret_key=self.APP_SECRET_KEY, payload={"exp": int(expire_time.timestamp()), "aid": app_id}) diff --git a/pkg/aloha/service/app.py b/pkg/aloha/service/app.py index b932903..4523fae 100644 --- a/pkg/aloha/service/app.py +++ b/pkg/aloha/service/app.py @@ -9,7 +9,7 @@ try: import uvloop - LOG.info("Using uvloop == %s for service event loop..." % uvloop.__version__) + LOG.info(f"Using uvloop == {uvloop.__version__} for service event loop...") asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()) except ImportError: LOG.info("[uvloop] NOT installed, fallback to asyncio loop! Consider `pip install uvloop`!") @@ -61,7 +61,7 @@ def start(self): LOG.info("Service interrupted by user") except Exception as e: LOG.error("Service error: %s", str(e)) - raise e + raise def stop(self): """Stop the server if it is currently running.""" diff --git a/pkg/aloha/service/http/base_api_client.py b/pkg/aloha/service/http/base_api_client.py index 1dd4648..e359b75 100644 --- a/pkg/aloha/service/http/base_api_client.py +++ b/pkg/aloha/service/http/base_api_client.py @@ -1,32 +1,31 @@ -"""Base HTTP client helpers for aloha API clients using httpx.""" - +import json import uuid from abc import ABC, abstractmethod from urllib.parse import urljoin -import httpx +import httpx2 from ...logger import LOG from ...settings import SETTINGS class AbstractApiClient(ABC): - """Common client behavior for aloha HTTP APIs using httpx.""" + """Common client behavior for aloha HTTP APIs using httpx2.""" LOG = LOG RETRY_METHOD_WHITELIST: frozenset = frozenset(["GET", "POST"]) RETRY_STATUS_FORCELIST: frozenset = frozenset({413, 429, 503, 502, 504}) config = SETTINGS.config - def __init__(self, url_endpoint: str = None, *args, **kwargs): + def __init__(self, url_endpoint: str | None = None, *args, **kwargs): """Store the endpoint used by the client.""" self.url_endpoint = url_endpoint or "" - LOG.debug("API Caller URL endpoint set to: %s" % self.url_endpoint) + LOG.debug(f"API Caller URL endpoint set to: {self.url_endpoint}") - def get_http_client(self, total_retries: int = 3, *args, **kwargs) -> httpx.AsyncClient: - """Create an httpx async client with retry support via custom transport.""" + def get_http_client(self, total_retries: int = 3, *args, **kwargs) -> httpx2.AsyncClient: + """Create an httpx2 async client with retry support via custom transport.""" # Create a custom transport that retries on specific status codes - from httpx import AsyncClient, Limits, Timeout + from httpx2 import AsyncClient, Limits, Timeout # Configure retry policy limits = Limits(max_keepalive_connections=20, max_connections=100, keepalive_expiry=30) @@ -55,12 +54,12 @@ def wrap_request_data(self, data: dict) -> dict: assert isinstance(data, dict), "Data object must be a dict!" raise NotImplementedError() - async def _async_call(self, api_url: str, data: dict = None, timeout: float = 5, **kwargs): + async def _async_call(self, api_url: str, data: dict | None = None, timeout: float = 5, **kwargs): """Async version: Call a remote API and return the parsed JSON response.""" - body = data or dict() + body = data or {} body.update(kwargs) payload = self.wrap_request_data(data=body) - LOG.debug("Calling api: %s" % api_url) + LOG.debug(f"Calling api: {api_url}") async with self.get_http_client() as client: resp = await client.post( @@ -69,13 +68,13 @@ async def _async_call(self, api_url: str, data: dict = None, timeout: float = 5, try: ret = resp.json() - except Exception as e: + except (json.JSONDecodeError, ValueError) as e: LOG.error(str(e)) raise RuntimeError(resp.text) return ret - def call(self, api_url: str, data: dict = None, timeout: float = 5, **kwargs): + def call(self, api_url: str, data: dict | None = None, timeout: float = 5, **kwargs): """Call a remote API and return the parsed JSON response (sync wrapper).""" import asyncio diff --git a/pkg/aloha/service/http/base_api_handler.py b/pkg/aloha/service/http/base_api_handler.py index bdfcdaa..4519bb3 100644 --- a/pkg/aloha/service/http/base_api_handler.py +++ b/pkg/aloha/service/http/base_api_handler.py @@ -4,8 +4,8 @@ import json import logging from abc import ABC -from datetime import datetime -from typing import Any, Dict, Optional +from datetime import datetime, timezone +from typing import Any, ClassVar from fastapi import APIRouter, Request, Response @@ -20,14 +20,14 @@ class AbstractApiHandler(ABC): """ LOG = LOG - MAP_ERROR_INFO: dict = {"BAD_REQUEST": {"code": "5101", "message": ["Bad request: fail to parse body as JSON object!"]}} + MAP_ERROR_INFO: ClassVar[dict] = {"BAD_REQUEST": {"code": "5101", "message": ["Bad request: fail to parse body as JSON object!"]}} def __init__(self): """Initialize request state used by subclasses.""" - self.api_args: Optional[tuple] = None - self.api_kwargs: Optional[dict] = None - self._request: Optional[Request] = None - self._response: Optional[Response] = None + self.api_args: tuple | None = None + self.api_kwargs: dict | None = None + self._request: Request | None = None + self._response: Response | None = None def response(self, *args, **kwargs) -> dict: """Subclasses must implement the business response.""" @@ -44,14 +44,14 @@ def request_header_content_type(self) -> str: def request_id(self) -> str: """Return or create a request identifier for tracing.""" if self._request is None: - return datetime.now().strftime("%Y%m%d-%H%M%S-%f") + return datetime.now(tz=timezone.utc).strftime("%Y%m%d-%H%M%S-%f") request_id = self._request.headers.get("Request-ID") if request_id is None: - request_id = datetime.now().strftime("%Y%m%d-%H%M%S-%f") + request_id = datetime.now(tz=timezone.utc).strftime("%Y%m%d-%H%M%S-%f") return request_id @property - def request_body(self) -> Optional[dict]: + def request_body(self) -> dict | None: """Parse the request body as JSON or multipart form data.""" content_type: str = self.request_header_content_type @@ -88,7 +88,7 @@ def request_param(self) -> dict: return ret - def get_request_files(self) -> Dict[str, list]: + def get_request_files(self) -> dict[str, list]: """Get uploaded files from multipart form data.""" if self._request is None: return {} @@ -106,11 +106,9 @@ def finish(self, data: Any, status_code: int = 200) -> Response: def set_header(self, key: str, value: str) -> None: """Set a response header (no-op in base class, overridden in FastAPI route).""" - pass - def set_status(self, status_code: int, reason: str = None) -> None: + def set_status(self, status_code: int, reason: str | None = None) -> None: """Set the response status code (no-op in base class).""" - pass async def _handle_request(self, request: Request, *args, **kwargs) -> Response: """Process the request and return a response.""" @@ -125,7 +123,7 @@ async def _handle_request(self, request: Request, *args, **kwargs) -> Response: return result except Exception as e: if self.LOG.level == logging.DEBUG: - self.LOG.error(e, exc_info=True) + self.LOG.exception("An internal error has occurred!") msgs = ["An internal error has occurred!", repr(e)] return self.finish({"code": 5201, "message": msgs}, status_code=500) diff --git a/pkg/aloha/service/http/files.py b/pkg/aloha/service/http/files.py index ed57a0c..086dd74 100644 --- a/pkg/aloha/service/http/files.py +++ b/pkg/aloha/service/http/files.py @@ -1,8 +1,8 @@ -"""Helpers for handling multipart upload files and remote file inputs using httpx.""" +"""Helpers for handling multipart upload files and remote file inputs using httpx2.""" import time -import httpx +import httpx2 from ...logger import LOG @@ -31,20 +31,16 @@ async def iter_over_request_files(request, url_files): # Handle files from URL for file_key, list_url in {"url_files": url_files or []}.items(): for url in sorted(set(list_url)): - try: - t_start = time.time() - async with httpx.AsyncClient(follow_redirects=True) as client: - resp = await client.get(url) - if resp.status_code == 200: - body = resp.content - content_type = resp.headers.get("Content-Type", "UNKNOWN") - else: - raise RuntimeError( - "Failed to download file after %s seconds with code=%s from URL %s" - % (time.time() - t_start, resp.status_code, url) - ) - except Exception as e: - raise e + t_start = time.time() + async with httpx2.AsyncClient(follow_redirects=True) as client: + resp = await client.get(url) + if resp.status_code == 200: + body = resp.content + content_type = resp.headers.get("Content-Type", "UNKNOWN") + else: + raise RuntimeError( + f"Failed to download file after {time.time() - t_start} seconds with code={resp.status_code} from URL {url}" + ) t_cost = time.time() - t_start LOG.info(f"File {url} has content type {content_type} and length bytes={len(body)}, downloaded in {t_cost} seconds") yield "url_files", url, content_type, body @@ -53,7 +49,7 @@ async def iter_over_request_files(request, url_files): def iter_over_request_files_sync(request, url_files): """Synchronous version of iter_over_request_files for backward compatibility. - This is a sync wrapper that uses httpx sync client. + This is a sync wrapper that uses httpx2 sync client. """ # Handle multipart uploaded files (from FastAPI form data) @@ -76,20 +72,16 @@ def iter_over_request_files_sync(request, url_files): # Handle files from URL for file_key, list_url in {"url_files": url_files or []}.items(): for url in sorted(set(list_url)): - try: - t_start = time.time() - with httpx.Client(follow_redirects=True) as client: - resp = client.get(url) - if resp.status_code == 200: - body = resp.content - content_type = resp.headers.get("Content-Type", "UNKNOWN") - else: - raise RuntimeError( - "Failed to download file after %s seconds with code=%s from URL %s" - % (time.time() - t_start, resp.status_code, url) - ) - except Exception as e: - raise e + t_start = time.time() + with httpx2.Client(follow_redirects=True) as client: + resp = client.get(url) + if resp.status_code == 200: + body = resp.content + content_type = resp.headers.get("Content-Type", "UNKNOWN") + else: + raise RuntimeError( + f"Failed to download file after {time.time() - t_start} seconds with code={resp.status_code} from URL {url}" + ) t_cost = time.time() - t_start LOG.info(f"File {url} has content type {content_type} and length bytes={len(body)}, downloaded in {t_cost} seconds") yield "url_files", url, content_type, body diff --git a/pkg/aloha/service/openapi/client.py b/pkg/aloha/service/openapi/client.py index a19c052..d124948 100644 --- a/pkg/aloha/service/openapi/client.py +++ b/pkg/aloha/service/openapi/client.py @@ -1,11 +1,8 @@ -"""Client helper for OpenAPI-style services protected by tokens.""" - import json -from datetime import datetime, timedelta -from typing import Optional +import time +from datetime import datetime, timedelta, timezone -from requests import Session -from requests.adapters import HTTPAdapter, Retry +import httpx2 from ...logger import LOG @@ -32,28 +29,43 @@ def __init__(self, url_oauth_get_token: str, client_id: str, client_secret: str, self.access_token = None @classmethod - def get_request_session(cls, total_retries: int = 10, *args, **kwargs) -> Session: - """Create a retry-enabled requests session.""" - session = Session() - # https://urllib3.readthedocs.io/en/latest/reference/urllib3.util.html#urllib3.util.Retry.DEFAULT_ALLOWED_METHODS - retries = Retry( - total=total_retries, - backoff_factor=0.1, - method_whitelist=cls.retry_method_whitelist, - status_forcelist=cls.retry_status_forcelist, - ) - for prefix in ("http://", "https://"): - session.mount(prefix, HTTPAdapter(max_retries=retries)) - return session + def get_request_session(cls, total_retries: int = 10, *args, **kwargs) -> httpx2.Client: + """Create an httpx2 client; retry policy is applied by ``_request``.""" + return httpx2.Client(*args, **kwargs) + + @classmethod + def _request(cls, method: str, url: str, total_retries: int = 10, **kwargs): + """Send a request with the former urllib3 retry policy.""" + if method.upper() not in cls.retry_method_whitelist: + return cls.get_request_session().request(method, url, **kwargs) + + last_error = None + for attempt in range(total_retries + 1): + client = cls.get_request_session() + try: + response = client.request(method, url, **kwargs) + if response.status_code not in cls.retry_status_forcelist or attempt == total_retries: + return response + except httpx2.HTTPError as error: + last_error = error + if attempt == total_retries: + raise + finally: + client.close() + time.sleep(0.1 * (2**attempt)) + + if last_error is not None: + raise last_error + raise RuntimeError("HTTP request failed without a response") def get_access_token(self) -> str: """Fetch or refresh the cached access token.""" - now = datetime.now() + now = datetime.now(tz=timezone.utc) if self.expires_at is None or self.expires_at > now: try: - # refresh access_token - resp = self.get_request_session().post( + resp = self._request( + "POST", self.url_oauth_get_token, timeout=5, json={"client_id": self.client_id, "client_secret": self.client_secret, "grant_type": self.grant_type}, @@ -61,56 +73,54 @@ def get_access_token(self) -> str: data = resp.json()["data"] if data is None or "access_token" not in data: - raise RuntimeError("Fail to fetch OpenAPI token with result: %s" % resp.text) + raise RuntimeError(f"Fail to fetch OpenAPI token with result: {resp.text}") self.access_token = data["access_token"] expires_in = int(data["expires_in"]) - self.expires_at = datetime.now() + timedelta(minutes=expires_in - 1) - except Exception as e: - msg = "Exception acquiring ESG access token from [%s]: %s" % (self.url_oauth_get_token, str(e)) + self.expires_at = datetime.now(tz=timezone.utc) + timedelta(minutes=expires_in - 1) + except Exception as e: # noqa: BLE001 + msg = f"Exception acquiring ESG access token from [{self.url_oauth_get_token}]: {e!s}" LOG.error(msg) return self.access_token def _get_request_url(self, url: str): """Attach access token and request id to the target URL.""" - request_url = "{url}?access_token={access_token}&request_id={request_id}".format( - url=url, access_token=self.get_access_token(), request_id=datetime.now().strftime("%Y%m%d-%H%M%S-%f") - ) + req_id = datetime.now(tz=timezone.utc).strftime("%Y%m%d-%H%M%S-%f") + request_url = f"{url}?access_token={self.get_access_token()}&request_id={req_id}" return request_url @staticmethod - def _get_data_from_esg_response(resp) -> Optional[dict]: + def _get_data_from_esg_response(resp) -> dict | None: """Parse a JSON response and unwrap legacy ESG payloads.""" try: return resp.json() - except (json.JSONDecodeError, JSONDecodeError): # requests may use `simplejson` + except (json.JSONDecodeError, JSONDecodeError): # simplejson may provide its own JSONDecodeError try: - # when data is wrapped by ESG content = resp.text.replace('"data":"', '"data":').replace('}"}', "}}") data = json.loads(content) return data.get("data", {}) except json.JSONDecodeError: - msg = "Cannot parse ESG response: %s" % resp.text + msg = f"Cannot parse ESG response: {resp.text}" raise ValueError(msg) - def post(self, url_api: str, body: dict, headers: dict = None, timeout: int = 5): + def post(self, url_api: str, body: dict, headers: dict | None = None, timeout: int = 5): """Send a POST request to the remote API.""" url = self._get_request_url(url_api) - LOG.debug("Calling ESG POST: %s" % url) + LOG.debug(f"Calling ESG POST: {url}") try: - resp = self.get_request_session().post(url=url, headers=headers, json=body, timeout=timeout) + resp = self._request("POST", url, headers=headers, json=body, timeout=timeout) return self._get_data_from_esg_response(resp) - except Exception as e: - LOG.error("Error calling ESG API POST [%s]: %s" % (url, str(e))) + except Exception as e: # noqa: BLE001 + LOG.error(f"Error calling ESG API POST [{url}]: {e!s}") - def get(self, url_api: str, body: dict, headers: dict = None, timeout: int = 5): + def get(self, url_api: str, body: dict, headers: dict | None = None, timeout: int = 5): """Send a GET request to the remote API.""" url = self._get_request_url(url_api) - LOG.debug("Calling ESG GET: %s" % url) + LOG.debug(f"Calling ESG GET: {url}") try: - resp = self.get_request_session().get(url=url, headers=headers, json=body, timeout=timeout) + resp = self._request("GET", url, headers=headers, json=body, timeout=timeout) return self._get_data_from_esg_response(resp) - except Exception as e: - LOG.error("Error calling ESG API GET [%s]: %s" % (url, str(e))) + except Exception as e: # noqa: BLE001 + LOG.error(f"Error calling ESG API GET [{url}]: {e!s}") diff --git a/pkg/aloha/service/web.py b/pkg/aloha/service/web.py index 048510f..057605d 100644 --- a/pkg/aloha/service/web.py +++ b/pkg/aloha/service/web.py @@ -1,9 +1,10 @@ """FastAPI web application assembly for aloha services.""" +import json import logging import os import re -from typing import Any, List, Tuple +from typing import Any from fastapi import FastAPI, Request from fastapi.responses import JSONResponse, Response @@ -15,11 +16,11 @@ setup_logger( logging.getLogger("uvicorn.access"), formatter_str="A> %(asctime)s> %(message)s", - module="access_%s" % (SETTINGS.config.get("APP_MODULE") or os.environ.get("APP_MODULE", "default")), + module=f"access_{SETTINGS.config.get('APP_MODULE') or os.environ.get('APP_MODULE', 'default')}", ) -def _load_routes(name: str) -> List[Tuple[str, Any]]: +def _load_routes(name: str) -> list[tuple[str, Any]]: """Load routes from a service module. Returns list of (url_pattern, handler_class) tuples. @@ -41,7 +42,7 @@ def _load_routes(name: str) -> List[Tuple[str, Any]]: class FastAPIApplication: """FastAPI application that loads routes from configured service modules.""" - def __init__(self, config: dict = None, **kwargs): + def __init__(self, config: dict | None = None, **kwargs): """Create the FastAPI application and its routes.""" self.config = config or {} self.app = FastAPI(title="Aloha Service", version="1.0.0", **kwargs) @@ -59,7 +60,7 @@ async def _default_404_handler(request: Request, exc: Exception): handler = handler_class(request=request) if hasattr(handler, "handle") and callable(handler.handle): return await handler.handle(request) - if hasattr(handler, "__call__") and callable(handler): + if callable(handler) and callable(handler): return await handler(request) if hasattr(handler, "response") and callable(handler.response): return await handler.response() @@ -77,15 +78,15 @@ def _setup_routes(self): routes = _load_routes(m) for url, handler_class in routes: self._register_handler(url, handler_class) - s_log_msg = "Loaded API module %-50s" % url + s_log_msg = f"Loaded API module {url:<50}" if LOG.level < logging.INFO: - s_log_msg += "\t from class %s" % str(handler_class) + s_log_msg += f"\t from class {handler_class!s}" LOG.info(s_log_msg) def _register_handler(self, url: str, handler_class): """Register a handler class as FastAPI routes based on its methods.""" - has_get = hasattr(handler_class, "get") and callable(getattr(handler_class, "get")) - has_post = hasattr(handler_class, "post") and callable(getattr(handler_class, "post")) + has_get = hasattr(handler_class, "get") and callable(handler_class.get) + has_post = hasattr(handler_class, "post") and callable(handler_class.post) # Determine path pattern for FastAPI fastapi_url, path_params = self._convert_url_pattern(url) @@ -110,7 +111,7 @@ async def post_handler(request: Request): try: body = await request.json() - except Exception: + except (json.JSONDecodeError, ValueError): body = {} kwargs.update(body) @@ -121,7 +122,7 @@ async def post_handler(request: Request): if isinstance(result, Response): return result # Otherwise, wrap in standard response format - resp = dict(code=5200, message=["success"]) + resp = {"code": 5200, "message": ["success"]} if isinstance(result, dict): resp["data"] = result.get("data", result) else: @@ -129,7 +130,7 @@ async def post_handler(request: Request): return JSONResponse(resp) except Exception as e: if handler.LOG.level == logging.DEBUG: - handler.LOG.error(e, exc_info=True) + handler.LOG.exception("Exception occurred during request processing") return JSONResponse({"code": 5201, "message": [repr(e)]}, status_code=500) self.app.post(fastapi_url)(post_handler) @@ -156,7 +157,7 @@ async def get_handler(request: Request): if isinstance(result, Response): return result # Otherwise, wrap in standard response format - resp = dict(code=5200, message=["success"]) + resp = {"code": 5200, "message": ["success"]} if isinstance(result, dict): resp["data"] = result.get("data", result) else: @@ -164,7 +165,7 @@ async def get_handler(request: Request): return JSONResponse(resp) except Exception as e: if handler.LOG.level == logging.DEBUG: - handler.LOG.error(e, exc_info=True) + handler.LOG.exception("Exception occurred during request processing") return JSONResponse({"code": 5201, "message": [repr(e)]}, status_code=500) self.app.get(fastapi_url)(get_handler) @@ -185,25 +186,25 @@ async def default_handler(request: Request): try: body = await request.json() - except Exception: + except (json.JSONDecodeError, ValueError): body = {} kwargs.update(body) - resp = dict(code=5200, message=["success"]) + resp = {"code": 5200, "message": ["success"]} try: result = handler.response(**kwargs) resp["data"] = result except Exception as e: if handler.LOG.level == logging.DEBUG: - handler.LOG.error(e, exc_info=True) + handler.LOG.exception("Exception occurred during request processing") return JSONResponse({"code": 5201, "message": [repr(e)]}, status_code=500) return JSONResponse(resp) self.app.post(fastapi_url)(default_handler) - def _convert_url_pattern(self, tornado_pattern: str) -> Tuple[str, bool]: + def _convert_url_pattern(self, tornado_pattern: str) -> tuple[str, bool]: """Convert Tornado URL pattern to FastAPI pattern. Tornado: /api/common/sys_info/(.*) @@ -228,7 +229,7 @@ def _match_path(self, tornado_pattern: str, path: str) -> dict: def get_port(self) -> int: """Get the configured port.""" service_settings = self.config.get("service", {}) - port = service_settings.get("port") or int(os.environ.get("PORT_SVC", 8000)) + port = service_settings.get("port") or int(os.environ.get("PORT_SVC", "8000")) port = int(os.environ.get("PORT", port)) return port diff --git a/pkg/aloha/settings.py b/pkg/aloha/settings.py index 20d12fc..31e5aec 100644 --- a/pkg/aloha/settings.py +++ b/pkg/aloha/settings.py @@ -61,14 +61,14 @@ def load_settings(self, config: Any) -> Any: :param config: The configuration data to load (dict or list). :return: The converted configuration object (AttrDict or list). - :raises ValueError: If the configuration data type is unsupported. + :raises TypeError: If the configuration data type is unsupported. """ if isinstance(config, dict): self._config = AttrDict({key: self.load_settings(value) for key, value in config.items()}) elif isinstance(config, list): self._config = [self.load_settings(value) for value in config] else: - raise ValueError("Unsupported config type: %s" % str(type(config))) + raise TypeError(f"Unsupported config type: {type(config)!s}") return self._config @property @@ -77,7 +77,7 @@ def config(self): Get the global configuration object. Lazily loads and parses configuration files on first access. It resolves active - HOCON configuration files based on the `FILES_CONFIG` or `ENV_PROFILE` environment + HOCON configuration files based on the `FILES_CONFIG` or `PROFILE_ENV` (or legacy `ENV_PROFILE`) environment variables, falls back to `main.conf` if not specified, and merges them into an `AttrDict`. :return: Merged global configuration settings as an AttrDict. diff --git a/pkg/aloha/testing/service_v1.py b/pkg/aloha/testing/service_v1.py index 2310b2a..56f5441 100644 --- a/pkg/aloha/testing/service_v1.py +++ b/pkg/aloha/testing/service_v1.py @@ -1,7 +1,7 @@ import json from abc import ABC -import requests +import httpx2 from aloha.service.api.v1 import APICaller @@ -20,13 +20,13 @@ def request_api(cls, api_url, timeout=5, **kwargs): """Class method to test an API call :param api_url: do NOT starts with slash (/) - :param timeout: requests timeout in seconds + :param timeout: httpx2 timeout in seconds :param kwargs: request data :return: """ payload = cls.wrap_request_data(data=kwargs) - url = "http://localhost:%s/%s" % (cls.api_url_base, api_url) - cls.LOG.debug("POST %s %s" % (url, json.dumps(payload, ensure_ascii=False, sort_keys=True))) - resp = requests.post(url, json=payload, timeout=timeout, headers={"Content-Type": "application/json"}).json() + url = f"http://localhost:{cls.api_url_base}/{api_url}" + cls.LOG.debug(f"POST {url} {json.dumps(payload, ensure_ascii=False, sort_keys=True)}") + resp = httpx2.post(url, json=payload, timeout=timeout, headers={"Content-Type": "application/json"}).json() cls.LOG.debug(resp) return resp diff --git a/pkg/aloha/testing/service_v2.py b/pkg/aloha/testing/service_v2.py index 2451e61..8dc2fb0 100644 --- a/pkg/aloha/testing/service_v2.py +++ b/pkg/aloha/testing/service_v2.py @@ -17,13 +17,13 @@ def request_api(cls, api_url, timeout=5, **kwargs): """Class method to test an API call :param api_url: do NOT start with slash (/) - :param timeout: requests timeout in seconds + :param timeout: httpx2 timeout in seconds :param kwargs: request data :return: """ - url = "http://localhost:%s/%s" % (cls.api_url_port, api_url) + url = f"http://localhost:{cls.api_url_port}/{api_url}" # cls.LOG.debug("POST %s %s" % (url, json.dumps(kwargs, ensure_ascii=False, sort_keys=True))) - # resp = requests.post( + # resp = httpx2.post( # url, json=kwargs, timeout=timeout, headers={'Content-Type': 'application/json'} # ).json() # cls.LOG.debug(resp) diff --git a/pkg/aloha/util/html.py b/pkg/aloha/util/html.py index 01d2ddb..953716d 100644 --- a/pkg/aloha/util/html.py +++ b/pkg/aloha/util/html.py @@ -12,7 +12,7 @@ def extract_img_url(string): for ii in html: images = ii.xpath("p/img/@src") return images[0] - except Exception as e: + except Exception as e: # noqa: BLE001 print(e, string) diff --git a/pkg/aloha/util/random.py b/pkg/aloha/util/random.py index 2ba4579..dc5dc37 100644 --- a/pkg/aloha/util/random.py +++ b/pkg/aloha/util/random.py @@ -8,9 +8,9 @@ "random_choice", "random_int", "random_ratio", - "random_uniform", "random_sample", "random_seed", + "random_uniform", ) random = SystemRandom() diff --git a/pkg/aloha/util/sys_cuda.py b/pkg/aloha/util/sys_cuda.py index 06a5dbd..4bc1acf 100644 --- a/pkg/aloha/util/sys_cuda.py +++ b/pkg/aloha/util/sys_cuda.py @@ -13,10 +13,10 @@ def get_gpu_status_for_tf(*args, **kwargs) -> dict: try: import tensorflow as tf - LOG.info("tensorflow version = %s" % tf.__version__) + LOG.info(f"tensorflow version = {tf.__version__}") status = Status(version=tf.__version__, gpu_availability=tf.test.is_gpu_available()) - except Exception as e: - msg = "Error detecting CUDA availability for tensorflow: %s" % str(e) + except Exception as e: # noqa: BLE001 + msg = f"Error detecting CUDA availability for tensorflow: {e!s}" LOG.warning(msg) return status._asdict() @@ -26,10 +26,10 @@ def get_gpu_status_for_torch(*args, **kwargs) -> dict: try: import torch - LOG.info("torch version = %s" % torch.__version__) + LOG.info(f"torch version = {torch.__version__}") status = Status(version=torch.__version__, gpu_availability=torch.cuda.is_available()) - except Exception as e: - msg = "Error detecting CUDA availability for torch: %s" % str(e) + except Exception as e: # noqa: BLE001 + msg = f"Error detecting CUDA availability for torch: {e!s}" LOG.warning(msg) return status._asdict() @@ -39,11 +39,11 @@ def get_gpu_status_for_paddle(*args, **kwargs) -> dict: try: import paddle - LOG.info("Paddlepaddle version = %s" % paddle.__version__) + LOG.info(f"Paddlepaddle version = {paddle.__version__}") paddle.utils.run_check() status = Status(version=paddle.__version__, gpu_availability=True) - except Exception as e: - msg = "Error detecting CUDA availability for paddle: %s" % str(e) + except Exception as e: # noqa: BLE001 + msg = f"Error detecting CUDA availability for paddle: {e!s}" LOG.warning(msg) return status._asdict() diff --git a/pkg/aloha/util/sys_gpu.py b/pkg/aloha/util/sys_gpu.py index b999031..a5003b2 100644 --- a/pkg/aloha/util/sys_gpu.py +++ b/pkg/aloha/util/sys_gpu.py @@ -8,7 +8,7 @@ import pynvml as nvml from pynvml.smi import nvidia_smi - LOG.debug("Using pynvml == %s" % nvml.__version__) + LOG.debug(f"Using pynvml == {nvml.__version__}") except ImportError: LOG.warn("Package `pynvml` NOT installed! Cannot get GPU info.") nvml = nvidia_smi = None @@ -27,7 +27,7 @@ def __init__(self): try: nvml.nvmlInit() LOG.debug("NVML loaded and initialized successfully.") - except Exception: + except Exception: # noqa: BLE001 LOG.error("Fail to initialize NVML!") nvml = None @@ -35,26 +35,26 @@ def __del__(self): try: if nvml is not None: nvml.nvmlShutdown() - except Exception as e: - LOG.error("Exception removing NvInfo: %s" % e) + except Exception as e: # noqa: BLE001 + LOG.error(f"Exception removing NvInfo: {e}") @staticmethod def get_driver_version() -> str: if nvml is not None: try: ver: bytes = nvml.nvmlSystemGetDriverVersion() - LOG.debug("GPU driver version %s" % str(ver)) + LOG.debug(f"GPU driver version {ver!s}") return ver.decode(encoding="UTF-8") - except Exception as e: - LOG.info("NVML library error: %s" % str(e)) + except Exception as e: # noqa: BLE001 + LOG.info(f"NVML library error: {e!s}") return "Unknown" @staticmethod def get_device_count() -> int: try: return nvml.nvmlDeviceGetCount() - except Exception as e: - LOG.info("NVML library error: %s" % str(e)) + except Exception as e: # noqa: BLE001 + LOG.info(f"NVML library error: {e!s}") return 0 def get_device_list(self) -> list: @@ -65,14 +65,14 @@ def get_device_list(self) -> list: try: name = nvml.nvmlDeviceGetName(handler).decode(encoding="UTF-8") - except Exception as e: - msg = "Failed to get device name: %s" % str(e) + except Exception as e: # noqa: BLE001 + msg = f"Failed to get device name: {e!s}" LOG.info(msg) try: arch = nvml.nvmlDeviceGetArchitecture(handler) - except Exception as e: - msg = "Failed to get device architecture: %s" % str(e) + except Exception as e: # noqa: BLE001 + msg = f"Failed to get device architecture: {e!s}" LOG.info(msg) device = Device(index=i, name=name, arch=arch) @@ -102,8 +102,8 @@ def get_smi(): return try: return nvidia_smi.getInstance() - except Exception as e: - LOG.warning("Failed to get smi: %s" % str(e)) + except Exception as e: # noqa: BLE001 + LOG.warning(f"Failed to get smi: {e!s}") return @@ -124,7 +124,7 @@ def get_gpu_info(*args, **kwargs) -> dict: smi = nv_info.get_smi() if smi is not None: if len(args) == 0: - args = "name;vbios_version;inforom.oem;compute-apps".split(";") + args = ["name", "vbios_version", "inforom.oem", "compute-apps"] for k in args: ret[k] = smi.DeviceQuery(k) diff --git a/pkg/aloha/util/sys_info.py b/pkg/aloha/util/sys_info.py index 05eafab..b2c7de6 100644 --- a/pkg/aloha/util/sys_info.py +++ b/pkg/aloha/util/sys_info.py @@ -1,5 +1,5 @@ import platform -from datetime import datetime +from datetime import datetime, timezone import psutil @@ -7,7 +7,7 @@ __all__ = ("get_sys_info",) -LOG.debug("Using psutil == %s" % psutil.__version__) +LOG.debug(f"Using psutil == {psutil.__version__}") def get_size(bytes, suffix="B"): @@ -37,7 +37,7 @@ def get_os_info(*args, **kwargs) -> dict: """ ret = {} - boot_time = datetime.fromtimestamp(psutil.boot_time()) + boot_time = datetime.fromtimestamp(psutil.boot_time(), tz=timezone.utc) ret["boot_time"] = boot_time.strftime("%Y-%m-%d %H:%M:%S.%f") uname = platform.uname() @@ -61,7 +61,7 @@ def get_cpu_info(*args, **kwargs) -> dict: "cpu_percent_total": f"{psutil.cpu_percent()}%", } for i, percentage in enumerate(psutil.cpu_percent(percpu=True, interval=1)): - ret["cpu_percent_core_%02d" % i] = f"{percentage}%" + ret[f"cpu_percent_core_{i:02d}"] = f"{percentage}%" return ret @@ -153,9 +153,9 @@ def get_net_info(*args, **kwargs) -> dict: family = str(address.family).split(".")[-1] family = {"AF_LINK": "mac", "AF_INET": "ipv4", "AF_INET6": "ipv6"}.get(family, family) - interface["%s_address" % family] = address.address - interface["%s_netmask" % family] = address.netmask - interface["%s_broadcast" % family] = address.broadcast + interface[f"{family}_address"] = address.address + interface[f"{family}_netmask"] = address.netmask + interface[f"{family}_broadcast"] = address.broadcast ret["interfaces"].append(interface) diff --git a/pkg/aloha/util/time.py b/pkg/aloha/util/time.py index 66f74bc..390db49 100644 --- a/pkg/aloha/util/time.py +++ b/pkg/aloha/util/time.py @@ -3,17 +3,18 @@ import asyncio import concurrent.futures import inspect -from typing import Any, Callable, Optional +from collections.abc import Callable +from typing import Any -__all__ = ("run_with_timeout", "run_async_with_timeout") +__all__ = ("run_async_with_timeout", "run_with_timeout") def run_with_timeout( func: Callable[..., Any], timeout_seconds: float, *args: Any, - fn_callback_success: Optional[Callable[[Any], Any]] = None, - fn_callback_fail: Optional[Callable[[Exception], Any]] = None, + fn_callback_success: Callable[[Any], Any] | None = None, + fn_callback_fail: Callable[[Exception], Any] | None = None, **kwargs: Any, ) -> Any: """Wrap a synchronous function call with a timeout. @@ -30,7 +31,7 @@ def run_with_timeout( if fn_callback_success is not None: fn_callback_success(result) return result - except Exception as e: + except Exception as e: # noqa: BLE001 if isinstance(e, concurrent.futures.TimeoutError): exc = TimeoutError(f"Operation timed out after {timeout_seconds} seconds") else: @@ -44,8 +45,8 @@ async def run_async_with_timeout( func: Callable[..., Any], timeout_seconds: float, *args: Any, - fn_callback_success: Optional[Callable[[Any], Any]] = None, - fn_callback_fail: Optional[Callable[[Exception], Any]] = None, + fn_callback_success: Callable[[Any], Any] | None = None, + fn_callback_fail: Callable[[Exception], Any] | None = None, **kwargs: Any, ) -> Any: """Wrap an asynchronous function call (coroutine function or sync function inside executor) with a timeout. @@ -67,7 +68,7 @@ async def run_async_with_timeout( if fn_callback_success is not None: fn_callback_success(result) return result - except Exception as e: + except Exception as e: # noqa: BLE001 if isinstance(e, (asyncio.TimeoutError, concurrent.futures.TimeoutError)): exc = TimeoutError(f"Operation timed out after {timeout_seconds} seconds") else: diff --git a/pkg/pyproject.toml b/pkg/pyproject.toml index 5032152..91b8135 100644 --- a/pkg/pyproject.toml +++ b/pkg/pyproject.toml @@ -37,7 +37,7 @@ aloha = "aloha.script.base:main" [project.optional-dependencies] build = ["Cython"] -service = ["psutil", "pyjwt", "fastapi", "httpx", "uvicorn"] +service = ["psutil", "pyjwt", "fastapi", "httpx2", "uvicorn"] db = [ "sqlalchemy", "psycopg[binary]", @@ -59,7 +59,7 @@ all = [ "psutil", "pyjwt", "fastapi", - "httpx", + "httpx2", "uvicorn", "sqlalchemy", "psycopg[binary]", diff --git a/pkg/setup.py b/pkg/setup.py index 650d860..e54f3ef 100644 --- a/pkg/setup.py +++ b/pkg/setup.py @@ -1,6 +1,6 @@ import os import shutil -from datetime import datetime +from datetime import datetime, timezone from setuptools import setup @@ -17,12 +17,12 @@ # 2. Dynamic Version Generation # Writes the version to aloha/_version.py using the current timestamp. -_t = datetime.now() -_version = "%s.%02d%02d.%02d%02d" % (_t.year, _t.month, _t.day, _t.hour, _t.minute) +_t = datetime.now(tz=timezone.utc) +_version = f"{_t.year}.{_t.month:02d}{_t.day:02d}.{_t.hour:02d}{_t.minute:02d}" version_file_path = os.path.join(base_dir, "aloha", "_version.py") with open(version_file_path, "wt") as fp: - fp.write('__version__ = "%s"\n' % _version) + fp.write(f'__version__ = "{_version}"\n') # 3. Trigger setup (reads configuration from pyproject.toml) setup() diff --git a/src/app_common/api/api_common_query_postgres.py b/src/app_common/api/api_common_query_postgres.py index c204f9f..1be1000 100644 --- a/src/app_common/api/api_common_query_postgres.py +++ b/src/app_common/api/api_common_query_postgres.py @@ -1,4 +1,3 @@ -from typing import Optional import pandas as pd from aloha.base import BaseModule @@ -9,7 +8,7 @@ class ApiQueryPostgres(APIHandler): - def response(self, sql: str, orient: str = "columns", config_profile: str = None, params=None, *args, **kwargs) -> str: + def response(self, sql: str, orient: str = "columns", config_profile: str | None = None, params=None, *args, **kwargs) -> str: op_query_db = QueryDb() df = op_query_db.query_db(sql=sql, config_profile=config_profile, params=params) ret = df.to_json(orient=orient, force_ascii=False) @@ -23,7 +22,7 @@ def get_operator(self, config_profile: str, *args, **kwargs): config_dict = self.config[config_profile] return PostgresOperator(config_dict) - def query_db(self, sql: str, config_profile: str = None, params=None, *args, **kwargs) -> Optional[pd.DataFrame]: + def query_db(self, sql: str, config_profile: str | None = None, params=None, *args, **kwargs) -> pd.DataFrame | None: op = self.get_operator(config_profile or "pg_rec_readonly") return pd.read_sql(sql=text(sql), con=op.engine, params=params) @@ -47,12 +46,12 @@ def main(): query = QueryDb() op = query.get_operator(**dict_params) - LOG.info("Connection string: %s" % op.connection_str) + LOG.info(f"Connection string: {op.connection_str}") if dict_params.get("sql", None) is not None: from tabulate import tabulate - LOG.info("Query result for: %s" % dict_params["sql"]) + LOG.info("Query result for: {}".format(dict_params["sql"])) df = query.query_db(**dict_params) table = tabulate(df, headers="keys", tablefmt="psql") print(table) diff --git a/src/app_common/api/api_common_sys_info.py b/src/app_common/api/api_common_sys_info.py index f3c0e45..90aa055 100644 --- a/src/app_common/api/api_common_sys_info.py +++ b/src/app_common/api/api_common_sys_info.py @@ -1,16 +1,16 @@ -from datetime import datetime +from datetime import datetime, timezone from aloha.service.api.v0 import APIHandler from aloha.util import sys_cuda, sys_gpu, sys_info def echo(*args, **kwargs): - return {"sys_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f"), **kwargs} + return {"sys_time": datetime.now(tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S.%f"), **kwargs} class SysStatusInfo(APIHandler): @staticmethod - def get_sys_info(kind: str = None, **kwargs) -> dict: + def get_sys_info(kind: str | None = None, **kwargs) -> dict: kinds = ["echo"] if kind is None or len(kind) == 0: pass @@ -39,15 +39,13 @@ def get_sys_info(kind: str = None, **kwargs) -> dict: return ret - def response(self, kind: str = None, *args, **kwargs) -> dict: + def response(self, kind: str | None = None, *args, **kwargs) -> dict: return self.get_sys_info(kind=kind) - async def get(self, kind: str = None, *args, **kwargs): + async def get(self, kind: str | None = None, *args, **kwargs): # Handle path_param from URL pattern - if "path_param" in kwargs: - # If kind is not set, try to use path_param as kind - if kind is None: - kind = kwargs.pop("path_param", None) + if "path_param" in kwargs and kind is None: + kind = kwargs.pop("path_param", None) data = self.get_sys_info(kind=kind) return self.finish(data) diff --git a/src/main.py b/src/main.py index 771972a..8d71bb5 100644 --- a/src/main.py +++ b/src/main.py @@ -8,11 +8,11 @@ if len(sys.argv) < 2: print(usage) - exit(-1) + sys.exit(-1) sys.argv.pop(0) m = importlib.import_module(sys.argv[0]) -f_main = getattr(m, "main") +f_main = m.main if f_main is None: print("Given module does not provides a `main()` function!") diff --git a/src/pyproject.toml b/src/pyproject.toml index 4eff00e..187a48f 100644 --- a/src/pyproject.toml +++ b/src/pyproject.toml @@ -21,7 +21,7 @@ dependencies = [ "psutil", "pyjwt", "fastapi", - "httpx", + "httpx2", "uvicorn", # Extras: db diff --git a/src/tests/test_paths.py b/src/tests/test_paths.py new file mode 100644 index 0000000..49f42ad --- /dev/null +++ b/src/tests/test_paths.py @@ -0,0 +1,120 @@ +import os +import unittest +import warnings +from unittest.mock import patch + +from aloha.config import paths + + +class TestConfigPaths(unittest.TestCase): + def setUp(self): + self.original_env = os.environ.copy() + for key in ("FILES_CONFIG", "PROFILE_ENV", "ENV_PROFILE", "DIR_CONFIG", "DIR_RESOURCE"): + os.environ.pop(key, None) + + def tearDown(self): + os.environ.clear() + os.environ.update(self.original_env) + + @patch("aloha.config.paths.os.path.exists", return_value=True) + def test_default_without_profile(self, mock_exists): + """When neither PROFILE_ENV nor ENV_PROFILE is set, default to main.conf without warning.""" + with warnings.catch_warnings(record=True) as captured_warnings: + warnings.simplefilter("always") + files = paths.get_config_files() + self.assertEqual(files, ["main.conf"]) + deprecation_warnings = [ + w for w in captured_warnings if issubclass(w.category, DeprecationWarning) + ] + self.assertEqual(len(deprecation_warnings), 0) + + @patch("aloha.config.paths.os.path.exists", return_value=True) + def test_profile_env_priority(self, mock_exists): + """When PROFILE_ENV is set, use main-{PROFILE_ENV}.conf without warning.""" + os.environ["PROFILE_ENV"] = "DEV" + with warnings.catch_warnings(record=True) as captured_warnings: + warnings.simplefilter("always") + files = paths.get_config_files() + self.assertEqual(files, ["main-DEV.conf"]) + deprecation_warnings = [ + w for w in captured_warnings if issubclass(w.category, DeprecationWarning) + ] + self.assertEqual(len(deprecation_warnings), 0) + + @patch("aloha.config.paths.os.path.exists", return_value=True) + def test_profile_env_precedence_over_env_profile(self, mock_exists): + """When both PROFILE_ENV and ENV_PROFILE are set, PROFILE_ENV takes precedence without warning.""" + os.environ["PROFILE_ENV"] = "PROD" + os.environ["ENV_PROFILE"] = "DEV" + with warnings.catch_warnings(record=True) as captured_warnings: + warnings.simplefilter("always") + files = paths.get_config_files() + self.assertEqual(files, ["main-PROD.conf"]) + deprecation_warnings = [ + w for w in captured_warnings if issubclass(w.category, DeprecationWarning) + ] + self.assertEqual(len(deprecation_warnings), 0) + + @patch("aloha.config.paths.os.path.exists", return_value=True) + def test_legacy_env_profile_fallback_with_warning(self, mock_exists): + """When PROFILE_ENV is unset but ENV_PROFILE is set, fall back to ENV_PROFILE and issue a DeprecationWarning.""" + os.environ["ENV_PROFILE"] = "STG" + with warnings.catch_warnings(record=True) as captured_warnings: + warnings.simplefilter("always") + files = paths.get_config_files() + self.assertEqual(files, ["main-STG.conf"]) + deprecation_warnings = [ + w for w in captured_warnings if issubclass(w.category, DeprecationWarning) + ] + self.assertEqual(len(deprecation_warnings), 1) + warning_msg = str(deprecation_warnings[0].message) + self.assertIn("ENV_PROFILE", warning_msg) + self.assertIn("deprecated", warning_msg) + self.assertIn("PROFILE_ENV", warning_msg) + + @patch("aloha.config.paths.os.path.exists", return_value=True) + def test_profile_env_empty_string_fallback(self, mock_exists): + """When PROFILE_ENV is empty string and ENV_PROFILE is set, fall back to ENV_PROFILE with warning.""" + os.environ["PROFILE_ENV"] = "" + os.environ["ENV_PROFILE"] = "STG" + with warnings.catch_warnings(record=True) as captured_warnings: + warnings.simplefilter("always") + files = paths.get_config_files() + self.assertEqual(files, ["main-STG.conf"]) + deprecation_warnings = [ + w for w in captured_warnings if issubclass(w.category, DeprecationWarning) + ] + self.assertEqual(len(deprecation_warnings), 1) + + @patch("aloha.config.paths.os.path.exists", return_value=True) + def test_both_empty_strings(self, mock_exists): + """When both PROFILE_ENV and ENV_PROFILE are empty strings, default to main.conf without warning.""" + os.environ["PROFILE_ENV"] = " " + os.environ["ENV_PROFILE"] = " " + with warnings.catch_warnings(record=True) as captured_warnings: + warnings.simplefilter("always") + files = paths.get_config_files() + self.assertEqual(files, ["main.conf"]) + deprecation_warnings = [ + w for w in captured_warnings if issubclass(w.category, DeprecationWarning) + ] + self.assertEqual(len(deprecation_warnings), 0) + + @patch("aloha.config.paths.os.path.exists", return_value=True) + def test_files_config_precedence(self, mock_exists): + """FILES_CONFIG overrides both PROFILE_ENV and ENV_PROFILE.""" + os.environ["FILES_CONFIG"] = "custom1.conf,custom2.conf" + os.environ["PROFILE_ENV"] = "PROD" + os.environ["ENV_PROFILE"] = "DEV" + with warnings.catch_warnings(record=True) as captured_warnings: + warnings.simplefilter("always") + files = paths.get_config_files() + self.assertEqual(files, ["custom1.conf", "custom2.conf"]) + deprecation_warnings = [ + w for w in captured_warnings if issubclass(w.category, DeprecationWarning) + ] + self.assertEqual(len(deprecation_warnings), 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/src/tests/test_time.py b/src/tests/test_time.py index 0d8581d..fab19dc 100644 --- a/src/tests/test_time.py +++ b/src/tests/test_time.py @@ -1,7 +1,9 @@ -import pytest -import time import asyncio -from aloha.util.time import run_with_timeout, run_async_with_timeout +import time + +import pytest +from aloha.util.time import run_async_with_timeout, run_with_timeout + # Helpers def sync_add(a, b, delay=0): @@ -9,14 +11,17 @@ def sync_add(a, b, delay=0): time.sleep(delay) return a + b + def sync_raise(): raise ValueError("sync error") + async def async_add(a, b, delay=0): if delay > 0: await asyncio.sleep(delay) return a + b + async def async_raise(): raise ValueError("async error") @@ -36,6 +41,7 @@ def on_success(res): assert success_called is True assert result_val == 5 + def test_sync_timeout(): fail_called = False error_val = None @@ -50,6 +56,7 @@ def on_fail(err): assert fail_called is True assert isinstance(error_val, TimeoutError) + def test_sync_exception(): fail_called = False error_val = None @@ -80,6 +87,7 @@ def on_success(res): assert success_called is True assert result_val == 5 + def test_async_timeout(): fail_called = False error_val = None @@ -94,6 +102,7 @@ def on_fail(err): assert fail_called is True assert isinstance(error_val, TimeoutError) + def test_async_exception(): fail_called = False error_val = None diff --git a/tool/cicd/docker-compose.app-demo.DEV.yml b/tool/cicd/docker-compose.app-demo.DEV.yml index 29a3d23..2e8cebf 100644 --- a/tool/cicd/docker-compose.app-demo.DEV.yml +++ b/tool/cicd/docker-compose.app-demo.DEV.yml @@ -14,7 +14,7 @@ services: restart: unless-stopped # env_file: ["../credentials/DEV-app-demo.env"] environment: - # - ENV_PROFILE=${ENV_PROFILE:-DEV} + # - PROFILE_ENV=${PROFILE_ENV:-DEV} - PROFILE_LOCALIZE=${PROFILE_LOCALIZE:-default} - PYTHONPATH=/root/app/pkg:/root/app/src:/root/app/notebook user: "0:0"