From 17ff4c9c0a97e8bc14fb600fefec1b99e8f9ab87 Mon Sep 17 00:00:00 2001 From: Junhao Liao Date: Mon, 29 Sep 2025 22:46:27 -0400 Subject: [PATCH 01/33] refactor(config): Use enum types for package storage and query engines and update serialization. --- .../clp-py-utils/clp_py_utils/clp_config.py | 34 +++++-------------- 1 file changed, 8 insertions(+), 26 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index 1ac4d7083f..58bff02a08 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -123,33 +123,9 @@ class AwsAuthType(LowercaseStrEnum): ec2 = auto() -VALID_STORAGE_ENGINES = [storage_engine.value for storage_engine in StorageEngine] -VALID_QUERY_ENGINES = [query_engine.value for query_engine in QueryEngine] - - class Package(BaseModel): - storage_engine: str = "clp" - query_engine: str = "clp" - - @field_validator("storage_engine") - @classmethod - def validate_storage_engine(cls, value): - if value not in VALID_STORAGE_ENGINES: - raise ValueError( - f"package.storage_engine must be one of the following" - f" {'|'.join(VALID_STORAGE_ENGINES)}" - ) - return value - - @field_validator("query_engine") - @classmethod - def validate_query_engine(cls, value): - if value not in VALID_QUERY_ENGINES: - raise ValueError( - f"package.query_engine must be one of the following" - f" {'|'.join(VALID_QUERY_ENGINES)}" - ) - return value + storage_engine: StorageEngine = StorageEngine.CLP + query_engine: QueryEngine = QueryEngine.CLP @model_validator(mode="after") def validate_query_engine_package_compatibility(self): @@ -173,6 +149,11 @@ def validate_query_engine_package_compatibility(self): return self + def dump_to_primitive_dict(self): + d = self.model_dump() + d["storage_engine"] = d["storage_engine"].value + d["query_engine"] = d["query_engine"].value + return d class Database(BaseModel): type: str = "mariadb" @@ -1035,6 +1016,7 @@ def get_runnable_components(self) -> Set[str]: def dump_to_primitive_dict(self): custom_serialized_fields = ( + "package", "database", "queue", "redis", From 4c014852a6588f614c6440682ec8a2cd02714448 Mon Sep 17 00:00:00 2001 From: Junhao Liao Date: Mon, 29 Sep 2025 22:48:31 -0400 Subject: [PATCH 02/33] refactor(config): Change custom_serialized_fields to a set and pass it directly to model_dump. --- components/clp-py-utils/clp_py_utils/clp_config.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index 58bff02a08..8b1dfc784b 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -1015,7 +1015,7 @@ def get_runnable_components(self) -> Set[str]: return ALL_COMPONENTS def dump_to_primitive_dict(self): - custom_serialized_fields = ( + custom_serialized_fields = { "package", "database", "queue", @@ -1023,8 +1023,8 @@ def dump_to_primitive_dict(self): "logs_input", "archive_output", "stream_output", - ) - d = self.model_dump(exclude=set(custom_serialized_fields)) + } + d = self.model_dump(exclude=custom_serialized_fields) for key in custom_serialized_fields: d[key] = getattr(self, key).dump_to_primitive_dict() From 3e484e20d4d3ade00a48c8288b91b9c342c3b85f Mon Sep 17 00:00:00 2001 From: Junhao Liao Date: Mon, 29 Sep 2025 23:04:35 -0400 Subject: [PATCH 03/33] refactor(config): Use shared annotated Port type for port fields and drop custom port validators. --- .../clp-py-utils/clp_py_utils/clp_config.py | 79 +++---------------- 1 file changed, 13 insertions(+), 66 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index 8b1dfc784b..131d6fe293 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -1,7 +1,7 @@ import os import pathlib from enum import auto -from typing import Any, Literal, Optional, Set, Union +from typing import Annotated, Any, Literal, Optional, Set, Union from dotenv import dotenv_values from pydantic import ( @@ -99,6 +99,9 @@ CLP_QUEUE_PASS_ENV_VAR_NAME = "CLP_QUEUE_PASS" CLP_REDIS_PASS_ENV_VAR_NAME = "CLP_REDIS_PASS" +# Types +Port = Annotated[int, Field(gt=0, lt=2**16)] + class StorageEngine(KebabCaseStrEnum): CLP = auto() @@ -155,10 +158,11 @@ def dump_to_primitive_dict(self): d["query_engine"] = d["query_engine"].value return d + class Database(BaseModel): type: str = "mariadb" host: str = "localhost" - port: int = 3306 + port: Port = 3306 name: str = "clp-db" ssl_cert: Optional[str] = None auto_commit: bool = False @@ -191,12 +195,6 @@ def validate_host(cls, value): raise ValueError("database.host cannot be empty.") return value - @field_validator("port") - @classmethod - def validate_port(cls, value): - _validate_port(cls, value) - return value - def ensure_credentials_loaded(self): if self.username is None or self.password is None: raise ValueError("Credentials not loaded.") @@ -281,15 +279,6 @@ def _validate_host(cls, value): raise ValueError(f"{cls.__name__}.host cannot be empty.") -def _validate_port(cls, value): - min_valid_port = 0 - max_valid_port = 2**16 - 1 - if min_valid_port > value or max_valid_port < value: - raise ValueError( - f"{cls.__name__}.port is not within valid range " f"{min_valid_port}-{max_valid_port}." - ) - - class CompressionScheduler(BaseModel): jobs_poll_delay: float = 0.1 # seconds logging_level: str = "INFO" @@ -303,7 +292,7 @@ def validate_logging_level(cls, value): class QueryScheduler(BaseModel): host: str = "localhost" - port: int = 7000 + port: Port = 7000 jobs_poll_delay: float = 0.1 # seconds num_archives_to_search_per_sub_job: int = 16 logging_level: str = "INFO" @@ -321,12 +310,6 @@ def validate_host(cls, value): raise ValueError(f"Cannot be empty.") return value - @field_validator("port") - @classmethod - def validate_port(cls, value): - _validate_port(cls, value) - return value - class CompressionWorker(BaseModel): logging_level: str = "INFO" @@ -350,7 +333,7 @@ def validate_logging_level(cls, value): class Redis(BaseModel): host: str = "localhost" - port: int = 6379 + port: Port = 6379 query_backend_database: int = 0 compression_backend_database: int = 1 # redis can perform authentication without a username @@ -363,12 +346,6 @@ def validate_host(cls, value): raise ValueError(f"{REDIS_COMPONENT_NAME}.host cannot be empty.") return value - @field_validator("port") - @classmethod - def validate_port(cls, value): - _validate_port(cls, value) - return value - def dump_to_primitive_dict(self): return self.model_dump(exclude={"password"}) @@ -392,7 +369,7 @@ def load_credentials_from_env(self): class Reducer(BaseModel): host: str = "localhost" - base_port: int = 14009 + base_port: Port = 14009 logging_level: str = "INFO" upsert_interval: int = 100 # milliseconds @@ -409,12 +386,6 @@ def validate_logging_level(cls, value): _validate_logging_level(cls, value) return value - @field_validator("base_port") - @classmethod - def validate_base_port(cls, value): - _validate_port(cls, value) - return value - @field_validator("upsert_interval") @classmethod def validate_upsert_interval(cls, value): @@ -425,7 +396,7 @@ def validate_upsert_interval(cls, value): class ResultsCache(BaseModel): host: str = "localhost" - port: int = 27017 + port: Port = 27017 db_name: str = "clp-query-results" stream_collection_name: str = "stream-files" retention_period: Optional[int] = 60 @@ -437,12 +408,6 @@ def validate_host(cls, value): raise ValueError(f"{RESULTS_CACHE_COMPONENT_NAME}.host cannot be empty.") return value - @field_validator("port") - @classmethod - def validate_port(cls, value): - _validate_port(cls, value) - return value - @field_validator("db_name") @classmethod def validate_db_name(cls, value): @@ -472,7 +437,7 @@ def get_uri(self): class Queue(BaseModel): host: str = "localhost" - port: int = 5672 + port: Port = 5672 username: Optional[str] = None password: Optional[str] = None @@ -484,12 +449,6 @@ def validate_host(cls, value): raise ValueError(f"{QUEUE_COMPONENT_NAME}.host cannot be empty.") return value - @field_validator("port") - @classmethod - def validate_port(cls, value): - _validate_port(cls, value) - return value - def dump_to_primitive_dict(self): return self.model_dump(exclude={"username", "password"}) @@ -781,7 +740,7 @@ def dump_to_primitive_dict(self): class WebUi(BaseModel): host: str = "localhost" - port: int = 4000 + port: Port = 4000 results_metadata_collection_name: str = "results-metadata" rate_limit: int = 1000 @@ -791,12 +750,6 @@ def validate_host(cls, value): _validate_host(cls, value) return value - @field_validator("port") - @classmethod - def validate_port(cls, value): - _validate_port(cls, value) - return value - @field_validator("results_metadata_collection_name") @classmethod def validate_results_metadata_collection_name(cls, value): @@ -834,7 +787,7 @@ def validate_logging_level(cls, value): class Presto(BaseModel): host: str - port: int + port: Port @field_validator("host") @classmethod @@ -842,12 +795,6 @@ def validate_host(cls, value): _validate_host(cls, value) return value - @field_validator("port") - @classmethod - def validate_port(cls, value): - _validate_port(cls, value) - return value - def _get_env_var(name: str) -> str: value = os.getenv(name) From d4419938e8976c2d6edd79949eacaba3a6d6c89a Mon Sep 17 00:00:00 2001 From: Junhao Liao Date: Mon, 29 Sep 2025 23:08:37 -0400 Subject: [PATCH 04/33] refactor(config): Consolidate host validation by introducing shared Host type and removing duplicated validators. --- .../clp-py-utils/clp_py_utils/clp_config.py | 76 +++---------------- 1 file changed, 9 insertions(+), 67 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index 131d6fe293..b04fc553ad 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -100,6 +100,7 @@ CLP_REDIS_PASS_ENV_VAR_NAME = "CLP_REDIS_PASS" # Types +Host = Annotated[str, Field(min_length=1)] Port = Annotated[int, Field(gt=0, lt=2**16)] @@ -161,7 +162,7 @@ def dump_to_primitive_dict(self): class Database(BaseModel): type: str = "mariadb" - host: str = "localhost" + host: Host = "localhost" port: Port = 3306 name: str = "clp-db" ssl_cert: Optional[str] = None @@ -188,13 +189,6 @@ def validate_name(cls, value): raise ValueError("database.name cannot be empty.") return value - @field_validator("host") - @classmethod - def validate_host(cls, value): - if "" == value: - raise ValueError("database.host cannot be empty.") - return value - def ensure_credentials_loaded(self): if self.username is None or self.password is None: raise ValueError("Credentials not loaded.") @@ -274,11 +268,6 @@ def _validate_logging_level(cls, value): ) -def _validate_host(cls, value): - if "" == value: - raise ValueError(f"{cls.__name__}.host cannot be empty.") - - class CompressionScheduler(BaseModel): jobs_poll_delay: float = 0.1 # seconds logging_level: str = "INFO" @@ -291,7 +280,7 @@ def validate_logging_level(cls, value): class QueryScheduler(BaseModel): - host: str = "localhost" + host: Host = "localhost" port: Port = 7000 jobs_poll_delay: float = 0.1 # seconds num_archives_to_search_per_sub_job: int = 16 @@ -303,13 +292,6 @@ def validate_logging_level(cls, value): _validate_logging_level(cls, value) return value - @field_validator("host") - @classmethod - def validate_host(cls, value): - if "" == value: - raise ValueError(f"Cannot be empty.") - return value - class CompressionWorker(BaseModel): logging_level: str = "INFO" @@ -332,20 +314,13 @@ def validate_logging_level(cls, value): class Redis(BaseModel): - host: str = "localhost" + host: Host = "localhost" port: Port = 6379 query_backend_database: int = 0 compression_backend_database: int = 1 # redis can perform authentication without a username password: Optional[str] = None - @field_validator("host") - @classmethod - def validate_host(cls, value): - if "" == value: - raise ValueError(f"{REDIS_COMPONENT_NAME}.host cannot be empty.") - return value - def dump_to_primitive_dict(self): return self.model_dump(exclude={"password"}) @@ -368,18 +343,11 @@ def load_credentials_from_env(self): class Reducer(BaseModel): - host: str = "localhost" + host: Host = "localhost" base_port: Port = 14009 logging_level: str = "INFO" upsert_interval: int = 100 # milliseconds - @field_validator("host") - @classmethod - def validate_host(cls, value): - if "" == value: - raise ValueError(f"{value} cannot be empty") - return value - @field_validator("logging_level") @classmethod def validate_logging_level(cls, value): @@ -395,19 +363,12 @@ def validate_upsert_interval(cls, value): class ResultsCache(BaseModel): - host: str = "localhost" + host: Host = "localhost" port: Port = 27017 db_name: str = "clp-query-results" stream_collection_name: str = "stream-files" retention_period: Optional[int] = 60 - @field_validator("host") - @classmethod - def validate_host(cls, value): - if "" == value: - raise ValueError(f"{RESULTS_CACHE_COMPONENT_NAME}.host cannot be empty.") - return value - @field_validator("db_name") @classmethod def validate_db_name(cls, value): @@ -436,19 +397,12 @@ def get_uri(self): class Queue(BaseModel): - host: str = "localhost" + host: Host = "localhost" port: Port = 5672 username: Optional[str] = None password: Optional[str] = None - @field_validator("host") - @classmethod - def validate_host(cls, value): - if "" == value: - raise ValueError(f"{QUEUE_COMPONENT_NAME}.host cannot be empty.") - return value - def dump_to_primitive_dict(self): return self.model_dump(exclude={"username", "password"}) @@ -739,17 +693,11 @@ def dump_to_primitive_dict(self): class WebUi(BaseModel): - host: str = "localhost" + host: Host = "localhost" port: Port = 4000 results_metadata_collection_name: str = "results-metadata" rate_limit: int = 1000 - @field_validator("host") - @classmethod - def validate_host(cls, value): - _validate_host(cls, value) - return value - @field_validator("results_metadata_collection_name") @classmethod def validate_results_metadata_collection_name(cls, value): @@ -786,15 +734,9 @@ def validate_logging_level(cls, value): class Presto(BaseModel): - host: str + host: Host port: Port - @field_validator("host") - @classmethod - def validate_host(cls, value): - _validate_host(cls, value) - return value - def _get_env_var(name: str) -> str: value = os.getenv(name) From 68a3a06093a9598fb4f1dca9d7588a3279f45272 Mon Sep 17 00:00:00 2001 From: Junhao Liao Date: Mon, 29 Sep 2025 23:17:16 -0400 Subject: [PATCH 05/33] refactor(config): Introduce DatabaseEngine enum, use it for database.type and adjust serialization accordingly. --- .../clp-py-utils/clp_py_utils/clp_config.py | 23 ++++++++----------- 1 file changed, 10 insertions(+), 13 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index b04fc553ad..f5a7ca2e59 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -109,6 +109,11 @@ class StorageEngine(KebabCaseStrEnum): CLP_S = auto() +class DatabaseEngine(KebabCaseStrEnum): + MARIADB = auto() + MYSQL = auto() + + class QueryEngine(KebabCaseStrEnum): CLP = auto() CLP_S = auto() @@ -161,7 +166,7 @@ def dump_to_primitive_dict(self): class Database(BaseModel): - type: str = "mariadb" + type: DatabaseEngine = DatabaseEngine.MARIADB host: Host = "localhost" port: Port = 3306 name: str = "clp-db" @@ -172,16 +177,6 @@ class Database(BaseModel): username: Optional[str] = None password: Optional[str] = None - @field_validator("type") - @classmethod - def validate_type(cls, value): - supported_database_types = ["mysql", "mariadb"] - if value not in supported_database_types: - raise ValueError( - f"database.type must be one of the following {'|'.join(supported_database_types)}" - ) - return value - @field_validator("name") @classmethod def validate_name(cls, value): @@ -223,7 +218,7 @@ def get_clp_connection_params_and_type(self, disable_localhost_socket_connection connection_params_and_type = { # NOTE: clp-core does not distinguish between mysql and mariadb - "type": "mysql", + "type": DatabaseEngine.MYSQL.value, "host": host, "port": self.port, "username": self.username, @@ -238,7 +233,9 @@ def get_clp_connection_params_and_type(self, disable_localhost_socket_connection return connection_params_and_type def dump_to_primitive_dict(self): - return self.model_dump(exclude={"username", "password"}) + d = self.model_dump(exclude={"username", "password"}) + d["type"] = d["type"].value + return d def load_credentials_from_file(self, credentials_file_path: pathlib.Path): config = read_yaml_config_file(credentials_file_path) From fbe91b0d19a4cbdd06b0a96304c6e94d21619e45 Mon Sep 17 00:00:00 2001 From: Junhao Liao Date: Mon, 29 Sep 2025 23:21:51 -0400 Subject: [PATCH 06/33] =?UTF-8?q?refactor(config):=20Replace=20manual=20no?= =?UTF-8?q?n=E2=80=91empty=20string=20validators=20with=20shared=20NonEmpt?= =?UTF-8?q?yStr=20type.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../clp-py-utils/clp_py_utils/clp_config.py | 79 +++---------------- 1 file changed, 10 insertions(+), 69 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index f5a7ca2e59..97758f631d 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -100,7 +100,8 @@ CLP_REDIS_PASS_ENV_VAR_NAME = "CLP_REDIS_PASS" # Types -Host = Annotated[str, Field(min_length=1)] +NonEmptyStr = Annotated[str, Field(min_length=1)] +Host = NonEmptyStr Port = Annotated[int, Field(gt=0, lt=2**16)] @@ -169,7 +170,7 @@ class Database(BaseModel): type: DatabaseEngine = DatabaseEngine.MARIADB host: Host = "localhost" port: Port = 3306 - name: str = "clp-db" + name: NonEmptyStr = "clp-db" ssl_cert: Optional[str] = None auto_commit: bool = False compress: bool = True @@ -177,13 +178,6 @@ class Database(BaseModel): username: Optional[str] = None password: Optional[str] = None - @field_validator("name") - @classmethod - def validate_name(cls, value): - if "" == value: - raise ValueError("database.name cannot be empty.") - return value - def ensure_credentials_loaded(self): if self.username is None or self.password is None: raise ValueError("Credentials not loaded.") @@ -362,26 +356,10 @@ def validate_upsert_interval(cls, value): class ResultsCache(BaseModel): host: Host = "localhost" port: Port = 27017 - db_name: str = "clp-query-results" - stream_collection_name: str = "stream-files" + db_name: NonEmptyStr = "clp-query-results" + stream_collection_name: NonEmptyStr = "stream-files" retention_period: Optional[int] = 60 - @field_validator("db_name") - @classmethod - def validate_db_name(cls, value): - if "" == value: - raise ValueError(f"{RESULTS_CACHE_COMPONENT_NAME}.db_name cannot be empty.") - return value - - @field_validator("stream_collection_name") - @classmethod - def validate_stream_collection_name(cls, value): - if "" == value: - raise ValueError( - f"{RESULTS_CACHE_COMPONENT_NAME}.stream_collection_name cannot be empty." - ) - return value - @field_validator("retention_period") @classmethod def validate_retention_period(cls, value): @@ -424,24 +402,10 @@ def load_credentials_from_env(self): class S3Credentials(BaseModel): - access_key_id: str - secret_access_key: str + access_key_id: NonEmptyStr + secret_access_key: NonEmptyStr session_token: Optional[str] = None - @field_validator("access_key_id") - @classmethod - def validate_access_key_id(cls, value): - if "" == value: - raise ValueError("access_key_id cannot be empty") - return value - - @field_validator("secret_access_key") - @classmethod - def validate_secret_access_key(cls, value): - if "" == value: - raise ValueError("secret_access_key cannot be empty") - return value - class AwsAuthentication(BaseModel): type: Literal[ @@ -482,25 +446,11 @@ def validate_authentication(cls, data): class S3Config(BaseModel): - region_code: str - bucket: str + region_code: NonEmptyStr + bucket: NonEmptyStr key_prefix: str aws_authentication: AwsAuthentication - @field_validator("region_code") - @classmethod - def validate_region_code(cls, value): - if "" == value: - raise ValueError("region_code cannot be empty") - return value - - @field_validator("bucket") - @classmethod - def validate_bucket(cls, value): - if "" == value: - raise ValueError("bucket cannot be empty") - return value - class S3IngestionConfig(BaseModel): type: Literal[StorageType.S3.value] = StorageType.S3.value @@ -692,18 +642,9 @@ def dump_to_primitive_dict(self): class WebUi(BaseModel): host: Host = "localhost" port: Port = 4000 - results_metadata_collection_name: str = "results-metadata" + results_metadata_collection_name: NonEmptyStr = "results-metadata" rate_limit: int = 1000 - @field_validator("results_metadata_collection_name") - @classmethod - def validate_results_metadata_collection_name(cls, value): - if "" == value: - raise ValueError( - f"{WEBUI_COMPONENT_NAME}.results_metadata_collection_name cannot be empty." - ) - return value - @field_validator("rate_limit") @classmethod def validate_rate_limit(cls, value): From e829a8615702d39225db266da5c366e42e5cf964 Mon Sep 17 00:00:00 2001 From: Junhao Liao Date: Mon, 29 Sep 2025 23:30:21 -0400 Subject: [PATCH 07/33] refactor(config): Use PositiveFloat type for jobs_poll_delay fields in scheduler configs. --- components/clp-py-utils/clp_py_utils/clp_config.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index 97758f631d..fbc4bc9e0f 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -100,6 +100,7 @@ CLP_REDIS_PASS_ENV_VAR_NAME = "CLP_REDIS_PASS" # Types +PositiveFloat = Annotated[float, Field(gt=0)] NonEmptyStr = Annotated[str, Field(min_length=1)] Host = NonEmptyStr Port = Annotated[int, Field(gt=0, lt=2**16)] @@ -260,7 +261,7 @@ def _validate_logging_level(cls, value): class CompressionScheduler(BaseModel): - jobs_poll_delay: float = 0.1 # seconds + jobs_poll_delay: PositiveFloat = 0.1 # seconds logging_level: str = "INFO" @field_validator("logging_level") @@ -273,7 +274,7 @@ def validate_logging_level(cls, value): class QueryScheduler(BaseModel): host: Host = "localhost" port: Port = 7000 - jobs_poll_delay: float = 0.1 # seconds + jobs_poll_delay: PositiveFloat = 0.1 # seconds num_archives_to_search_per_sub_job: int = 16 logging_level: str = "INFO" From 79b80433e9ab6672f790ae09a98422453f4072c8 Mon Sep 17 00:00:00 2001 From: Junhao Liao Date: Mon, 29 Sep 2025 23:31:26 -0400 Subject: [PATCH 08/33] Refactor(config): Rename jobs_poll_delay to jobs_poll_delay_sec in scheduler configs and update references. --- components/clp-py-utils/clp_py_utils/clp_config.py | 4 ++-- .../scheduler/compress/compression_scheduler.py | 2 +- .../job_orchestration/scheduler/query/query_scheduler.py | 4 ++-- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index fbc4bc9e0f..4b45ba191b 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -261,7 +261,7 @@ def _validate_logging_level(cls, value): class CompressionScheduler(BaseModel): - jobs_poll_delay: PositiveFloat = 0.1 # seconds + jobs_poll_delay_sec: PositiveFloat = 0.1 logging_level: str = "INFO" @field_validator("logging_level") @@ -274,7 +274,7 @@ def validate_logging_level(cls, value): class QueryScheduler(BaseModel): host: Host = "localhost" port: Port = 7000 - jobs_poll_delay: PositiveFloat = 0.1 # seconds + jobs_poll_delay_sec: PositiveFloat = 0.1 num_archives_to_search_per_sub_job: int = 16 logging_level: str = "INFO" diff --git a/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py b/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py index a12a0cf6a6..4613b72276 100644 --- a/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py +++ b/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py @@ -523,7 +523,7 @@ def main(argv): clp_metadata_db_connection_config, ) poll_running_jobs(db_conn, db_cursor) - time.sleep(clp_config.compression_scheduler.jobs_poll_delay) + time.sleep(clp_config.compression_scheduler.jobs_poll_delay_sec) except KeyboardInterrupt: logger.info("Forcefully shutting down") return -1 diff --git a/components/job-orchestration/job_orchestration/scheduler/query/query_scheduler.py b/components/job-orchestration/job_orchestration/scheduler/query/query_scheduler.py index 7e73572725..417bbecdf1 100644 --- a/components/job-orchestration/job_orchestration/scheduler/query/query_scheduler.py +++ b/components/job-orchestration/job_orchestration/scheduler/query/query_scheduler.py @@ -1180,7 +1180,7 @@ async def main(argv: List[str]) -> int: logger.exception("Failed to kill hanging query jobs.") return -1 - logger.debug(f"Job polling interval {clp_config.query_scheduler.jobs_poll_delay} seconds.") + logger.debug(f"Job polling interval {clp_config.query_scheduler.jobs_poll_delay_sec} seconds.") try: reducer_handler = await asyncio.start_server( lambda reader, writer: handle_reducer_connection( @@ -1214,7 +1214,7 @@ async def main(argv: List[str]) -> int: ), results_cache_uri=clp_config.results_cache.get_uri(), stream_collection_name=clp_config.results_cache.stream_collection_name, - jobs_poll_delay=clp_config.query_scheduler.jobs_poll_delay, + jobs_poll_delay=clp_config.query_scheduler.jobs_poll_delay_sec, num_archives_to_search_per_sub_job=batch_size, archive_retention_period=clp_config.archive_output.retention_period, ) From 354b019fb3f1ec592ed43d74cc0332d100791248 Mon Sep 17 00:00:00 2001 From: Junhao Liao Date: Mon, 29 Sep 2025 23:37:03 -0400 Subject: [PATCH 09/33] refactor(config): Replace int fields with PositiveInt and remove redundant validators. --- .../clp-py-utils/clp_py_utils/clp_config.py | 93 ++++--------------- 1 file changed, 16 insertions(+), 77 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index 4b45ba191b..e3c8b7730d 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -99,9 +99,11 @@ CLP_QUEUE_PASS_ENV_VAR_NAME = "CLP_QUEUE_PASS" CLP_REDIS_PASS_ENV_VAR_NAME = "CLP_REDIS_PASS" -# Types -PositiveFloat = Annotated[float, Field(gt=0)] +# Generic types NonEmptyStr = Annotated[str, Field(min_length=1)] +PositiveFloat = Annotated[float, Field(gt=0)] +PositiveInt = Annotated[int, Field(gt=0)] +# Type aliases Host = NonEmptyStr Port = Annotated[int, Field(gt=0, lt=2**16)] @@ -275,7 +277,7 @@ class QueryScheduler(BaseModel): host: Host = "localhost" port: Port = 7000 jobs_poll_delay_sec: PositiveFloat = 0.1 - num_archives_to_search_per_sub_job: int = 16 + num_archives_to_search_per_sub_job: PositiveInt = 16 logging_level: str = "INFO" @field_validator("logging_level") @@ -338,7 +340,7 @@ class Reducer(BaseModel): host: Host = "localhost" base_port: Port = 14009 logging_level: str = "INFO" - upsert_interval: int = 100 # milliseconds + upsert_interval: PositiveInt = 100 # milliseconds @field_validator("logging_level") @classmethod @@ -346,27 +348,13 @@ def validate_logging_level(cls, value): _validate_logging_level(cls, value) return value - @field_validator("upsert_interval") - @classmethod - def validate_upsert_interval(cls, value): - if not value > 0: - raise ValueError(f"{value} is not greater than zero") - return value - class ResultsCache(BaseModel): host: Host = "localhost" port: Port = 27017 db_name: NonEmptyStr = "clp-query-results" stream_collection_name: NonEmptyStr = "stream-files" - retention_period: Optional[int] = 60 - - @field_validator("retention_period") - @classmethod - def validate_retention_period(cls, value): - if value is not None and value <= 0: - raise ValueError("retention_period must be greater than 0") - return value + retention_period: Optional[PositiveInt] = 60 def get_uri(self): return f"mongodb://{self.host}:{self.port}/{self.db_name}" @@ -556,40 +544,12 @@ def _set_directory_for_storage_config( class ArchiveOutput(BaseModel): storage: Union[ArchiveFsStorage, ArchiveS3Storage] = ArchiveFsStorage() - target_archive_size: int = 256 * 1024 * 1024 # 256 MB - target_dictionaries_size: int = 32 * 1024 * 1024 # 32 MB - target_encoded_file_size: int = 256 * 1024 * 1024 # 256 MB - target_segment_size: int = 256 * 1024 * 1024 # 256 MB + target_archive_size: PositiveInt = 256 * 1024 * 1024 # 256 MB + target_dictionaries_size: PositiveInt = 32 * 1024 * 1024 # 32 MB + target_encoded_file_size: PositiveInt = 256 * 1024 * 1024 # 256 MB + target_segment_size: PositiveInt = 256 * 1024 * 1024 # 256 MB compression_level: int = 3 - retention_period: Optional[int] = None - - @field_validator("target_archive_size") - @classmethod - def validate_target_archive_size(cls, value): - if value <= 0: - raise ValueError("target_archive_size must be greater than 0") - return value - - @field_validator("target_dictionaries_size") - @classmethod - def validate_target_dictionaries_size(cls, value): - if value <= 0: - raise ValueError("target_dictionaries_size must be greater than 0") - return value - - @field_validator("target_encoded_file_size") - @classmethod - def validate_target_encoded_file_size(cls, value): - if value <= 0: - raise ValueError("target_encoded_file_size must be greater than 0") - return value - - @field_validator("target_segment_size") - @classmethod - def validate_target_segment_size(cls, value): - if value <= 0: - raise ValueError("target_segment_size must be greater than 0") - return value + retention_period: Optional[PositiveInt] = None @field_validator("compression_level") @classmethod @@ -598,13 +558,6 @@ def validate_compression_level(cls, value): raise ValueError("compression_level must be a value from 1 to 19") return value - @field_validator("retention_period") - @classmethod - def validate_retention_period(cls, value): - if value is not None and value <= 0: - raise ValueError("retention_period must be greater than 0") - return value - def set_directory(self, directory: pathlib.Path): _set_directory_for_storage_config(self.storage, directory) @@ -619,14 +572,7 @@ def dump_to_primitive_dict(self): class StreamOutput(BaseModel): storage: Union[StreamFsStorage, StreamS3Storage] = StreamFsStorage() - target_uncompressed_size: int = 128 * 1024 * 1024 - - @field_validator("target_uncompressed_size") - @classmethod - def validate_target_uncompressed_size(cls, value): - if value <= 0: - raise ValueError("target_uncompressed_size must be greater than 0") - return value + target_uncompressed_size: PositiveInt = 128 * 1024 * 1024 def set_directory(self, directory: pathlib.Path): _set_directory_for_storage_config(self.storage, directory) @@ -644,21 +590,14 @@ class WebUi(BaseModel): host: Host = "localhost" port: Port = 4000 results_metadata_collection_name: NonEmptyStr = "results-metadata" - rate_limit: int = 1000 - - @field_validator("rate_limit") - @classmethod - def validate_rate_limit(cls, value): - if value <= 0: - raise ValueError(f"rate_limit must be greater than 0") - return value + rate_limit: PositiveInt = 1000 class SweepInterval(BaseModel): model_config = ConfigDict(extra="forbid") - archive: int = Field(default=60, gt=0) - search_result: int = Field(default=30, gt=0) + archive: PositiveInt = 60 + search_result: PositiveInt = 30 class GarbageCollector(BaseModel): From e60b39a6bfbab3133577b67fb8da3b4e41d69105 Mon Sep 17 00:00:00 2001 From: Junhao Liao Date: Mon, 29 Sep 2025 23:41:53 -0400 Subject: [PATCH 10/33] refactor(config): Update optional string fields to use NonEmptyStr type. --- components/clp-py-utils/clp_py_utils/clp_config.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index e3c8b7730d..b7de55246b 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -174,7 +174,7 @@ class Database(BaseModel): host: Host = "localhost" port: Port = 3306 name: NonEmptyStr = "clp-db" - ssl_cert: Optional[str] = None + ssl_cert: Optional[NonEmptyStr] = None auto_commit: bool = False compress: bool = True @@ -364,7 +364,7 @@ class Queue(BaseModel): host: Host = "localhost" port: Port = 5672 - username: Optional[str] = None + username: Optional[NonEmptyStr] = None password: Optional[str] = None def dump_to_primitive_dict(self): @@ -393,7 +393,7 @@ def load_credentials_from_env(self): class S3Credentials(BaseModel): access_key_id: NonEmptyStr secret_access_key: NonEmptyStr - session_token: Optional[str] = None + session_token: Optional[NonEmptyStr] = None class AwsAuthentication(BaseModel): @@ -403,7 +403,7 @@ class AwsAuthentication(BaseModel): AwsAuthType.env_vars.value, AwsAuthType.ec2.value, ] - profile: Optional[str] = None + profile: Optional[NonEmptyStr] = None credentials: Optional[S3Credentials] = None @model_validator(mode="before") @@ -624,7 +624,7 @@ def _get_env_var(name: str) -> str: class CLPConfig(BaseModel): - execution_container: Optional[str] = None + execution_container: Optional[NonEmptyStr] = None logs_input: Union[FsIngestionConfig, S3IngestionConfig] = FsIngestionConfig() From 3372a4be33edcf420eab72dd63d38f119a44fc03 Mon Sep 17 00:00:00 2001 From: Junhao Liao Date: Mon, 29 Sep 2025 23:46:39 -0400 Subject: [PATCH 11/33] docs(config): Update comment to specify specific types. --- components/clp-py-utils/clp_py_utils/clp_config.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index b7de55246b..bd0afe8cab 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -103,7 +103,7 @@ NonEmptyStr = Annotated[str, Field(min_length=1)] PositiveFloat = Annotated[float, Field(gt=0)] PositiveInt = Annotated[int, Field(gt=0)] -# Type aliases +# Specific types Host = NonEmptyStr Port = Annotated[int, Field(gt=0, lt=2**16)] From 68740f5558d76e10742cd4f93874584ea0ebb873 Mon Sep 17 00:00:00 2001 From: Junhao Liao Date: Mon, 29 Sep 2025 23:47:27 -0400 Subject: [PATCH 12/33] refactor(config): Use ZstdCompressionLevel for compression_level and remove its validator. --- components/clp-py-utils/clp_py_utils/clp_config.py | 10 ++-------- 1 file changed, 2 insertions(+), 8 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index bd0afe8cab..d36157ea05 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -106,6 +106,7 @@ # Specific types Host = NonEmptyStr Port = Annotated[int, Field(gt=0, lt=2**16)] +ZstdCompressionLevel = Annotated[int, Field(ge=1, le=19)] class StorageEngine(KebabCaseStrEnum): @@ -548,16 +549,9 @@ class ArchiveOutput(BaseModel): target_dictionaries_size: PositiveInt = 32 * 1024 * 1024 # 32 MB target_encoded_file_size: PositiveInt = 256 * 1024 * 1024 # 256 MB target_segment_size: PositiveInt = 256 * 1024 * 1024 # 256 MB - compression_level: int = 3 + compression_level: ZstdCompressionLevel = 3 retention_period: Optional[PositiveInt] = None - @field_validator("compression_level") - @classmethod - def validate_compression_level(cls, value): - if value < 1 or value > 19: - raise ValueError("compression_level must be a value from 1 to 19") - return value - def set_directory(self, directory: pathlib.Path): _set_directory_for_storage_config(self.storage, directory) From 420f30f58f16a9d6eb2257e3738b26b4f8dd4788 Mon Sep 17 00:00:00 2001 From: Junhao Liao Date: Tue, 30 Sep 2025 02:21:30 -0400 Subject: [PATCH 13/33] refactor(config): Replace string logging_level fields with LoggingLevel type and remove validation helpers. --- .../clp-py-utils/clp_py_utils/clp_config.py | 58 +++---------------- .../clp-py-utils/clp_py_utils/clp_logging.py | 30 ++++------ 2 files changed, 19 insertions(+), 69 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index d36157ea05..546096a173 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -14,7 +14,7 @@ ) from strenum import KebabCaseStrEnum, LowercaseStrEnum -from .clp_logging import get_valid_logging_level, is_valid_logging_level +from .clp_logging import LoggingLevel from .core import ( get_config_value, make_config_path_absolute, @@ -255,23 +255,9 @@ def load_credentials_from_env(self): self.password = _get_env_var(CLP_DB_PASS_ENV_VAR_NAME) -def _validate_logging_level(cls, value): - if not is_valid_logging_level(value): - raise ValueError( - f"{cls.__name__}: '{value}' is not a valid logging level. Use one of" - f" {get_valid_logging_level()}" - ) - - class CompressionScheduler(BaseModel): jobs_poll_delay_sec: PositiveFloat = 0.1 - logging_level: str = "INFO" - - @field_validator("logging_level") - @classmethod - def validate_logging_level(cls, value): - _validate_logging_level(cls, value) - return value + logging_level: LoggingLevel = "INFO" class QueryScheduler(BaseModel): @@ -279,33 +265,15 @@ class QueryScheduler(BaseModel): port: Port = 7000 jobs_poll_delay_sec: PositiveFloat = 0.1 num_archives_to_search_per_sub_job: PositiveInt = 16 - logging_level: str = "INFO" - - @field_validator("logging_level") - @classmethod - def validate_logging_level(cls, value): - _validate_logging_level(cls, value) - return value + logging_level: LoggingLevel = "INFO" class CompressionWorker(BaseModel): - logging_level: str = "INFO" - - @field_validator("logging_level") - @classmethod - def validate_logging_level(cls, value): - _validate_logging_level(cls, value) - return value + logging_level: LoggingLevel = "INFO" class QueryWorker(BaseModel): - logging_level: str = "INFO" - - @field_validator("logging_level") - @classmethod - def validate_logging_level(cls, value): - _validate_logging_level(cls, value) - return value + logging_level: LoggingLevel = "INFO" class Redis(BaseModel): @@ -340,15 +308,9 @@ def load_credentials_from_env(self): class Reducer(BaseModel): host: Host = "localhost" base_port: Port = 14009 - logging_level: str = "INFO" + logging_level: LoggingLevel = "INFO" upsert_interval: PositiveInt = 100 # milliseconds - @field_validator("logging_level") - @classmethod - def validate_logging_level(cls, value): - _validate_logging_level(cls, value) - return value - class ResultsCache(BaseModel): host: Host = "localhost" @@ -595,15 +557,9 @@ class SweepInterval(BaseModel): class GarbageCollector(BaseModel): - logging_level: str = "INFO" + logging_level: LoggingLevel = "INFO" sweep_interval: SweepInterval = SweepInterval() - @field_validator("logging_level") - @classmethod - def validate_logging_level(cls, value): - _validate_logging_level(cls, value) - return value - class Presto(BaseModel): host: Host diff --git a/components/clp-py-utils/clp_py_utils/clp_logging.py b/components/clp-py-utils/clp_py_utils/clp_logging.py index dfe2ae4d8e..23d58602a2 100644 --- a/components/clp-py-utils/clp_py_utils/clp_logging.py +++ b/components/clp-py-utils/clp_py_utils/clp_logging.py @@ -1,13 +1,14 @@ import logging +from typing import get_args, Literal -LOGGING_LEVEL_MAPPING = { - "INFO": logging.INFO, - "DEBUG": logging.DEBUG, - "WARN": logging.WARNING, - "WARNING": logging.WARNING, - "ERROR": logging.ERROR, - "CRITICAL": logging.CRITICAL, -} +LoggingLevel = Literal[ + "INFO", + "DEBUG", + "WARN", + "WARNING", + "ERROR", + "CRITICAL", +] def get_logging_formatter(): @@ -25,17 +26,10 @@ def get_logger(name: str): return logger -def get_valid_logging_level(): - return [i for i in LOGGING_LEVEL_MAPPING.keys()] - - -def is_valid_logging_level(level: str): - return level in LOGGING_LEVEL_MAPPING - - def set_logging_level(logger: logging.Logger, level: str): - if not is_valid_logging_level(level): + if level not in get_args(LoggingLevel): logger.warning(f"Invalid logging level: {level}, using INFO as default") logger.setLevel(logging.INFO) return - logger.setLevel(LOGGING_LEVEL_MAPPING[level]) + + logger.setLevel(level) From 6a584e016e984f98e36c8fc8bc28cf2cb6846f44 Mon Sep 17 00:00:00 2001 From: Junhao Liao Date: Tue, 30 Sep 2025 02:41:22 -0400 Subject: [PATCH 14/33] revert jobs_poll_delay rename --- components/clp-py-utils/clp_py_utils/clp_config.py | 4 ++-- .../scheduler/compress/compression_scheduler.py | 2 +- .../job_orchestration/scheduler/query/query_scheduler.py | 4 ++-- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index 546096a173..b6743d7f94 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -256,14 +256,14 @@ def load_credentials_from_env(self): class CompressionScheduler(BaseModel): - jobs_poll_delay_sec: PositiveFloat = 0.1 + jobs_poll_delay: PositiveFloat = 0.1 # seconds logging_level: LoggingLevel = "INFO" class QueryScheduler(BaseModel): host: Host = "localhost" port: Port = 7000 - jobs_poll_delay_sec: PositiveFloat = 0.1 + jobs_poll_delay: PositiveFloat = 0.1 # seconds num_archives_to_search_per_sub_job: PositiveInt = 16 logging_level: LoggingLevel = "INFO" diff --git a/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py b/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py index 4613b72276..a12a0cf6a6 100644 --- a/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py +++ b/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py @@ -523,7 +523,7 @@ def main(argv): clp_metadata_db_connection_config, ) poll_running_jobs(db_conn, db_cursor) - time.sleep(clp_config.compression_scheduler.jobs_poll_delay_sec) + time.sleep(clp_config.compression_scheduler.jobs_poll_delay) except KeyboardInterrupt: logger.info("Forcefully shutting down") return -1 diff --git a/components/job-orchestration/job_orchestration/scheduler/query/query_scheduler.py b/components/job-orchestration/job_orchestration/scheduler/query/query_scheduler.py index 417bbecdf1..7e73572725 100644 --- a/components/job-orchestration/job_orchestration/scheduler/query/query_scheduler.py +++ b/components/job-orchestration/job_orchestration/scheduler/query/query_scheduler.py @@ -1180,7 +1180,7 @@ async def main(argv: List[str]) -> int: logger.exception("Failed to kill hanging query jobs.") return -1 - logger.debug(f"Job polling interval {clp_config.query_scheduler.jobs_poll_delay_sec} seconds.") + logger.debug(f"Job polling interval {clp_config.query_scheduler.jobs_poll_delay} seconds.") try: reducer_handler = await asyncio.start_server( lambda reader, writer: handle_reducer_connection( @@ -1214,7 +1214,7 @@ async def main(argv: List[str]) -> int: ), results_cache_uri=clp_config.results_cache.get_uri(), stream_collection_name=clp_config.results_cache.stream_collection_name, - jobs_poll_delay=clp_config.query_scheduler.jobs_poll_delay_sec, + jobs_poll_delay=clp_config.query_scheduler.jobs_poll_delay, num_archives_to_search_per_sub_job=batch_size, archive_retention_period=clp_config.archive_output.retention_period, ) From cc9a1f4384ca898891deec1f8c739ed05f44b608 Mon Sep 17 00:00:00 2001 From: Junhao Liao Date: Fri, 10 Oct 2025 03:18:04 -0400 Subject: [PATCH 15/33] refactor(config): Rename type `Host` -> `DomainStr`; Add TODO docstring about plan for replacement. --- .../clp-py-utils/clp_py_utils/clp_config.py | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index b6743d7f94..7c8cd9feb4 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -104,7 +104,8 @@ PositiveFloat = Annotated[float, Field(gt=0)] PositiveInt = Annotated[int, Field(gt=0)] # Specific types -Host = NonEmptyStr +# TODO: Replace this with pydantic_extra_types.domain.DomainStr. +DomainStr = NonEmptyStr Port = Annotated[int, Field(gt=0, lt=2**16)] ZstdCompressionLevel = Annotated[int, Field(ge=1, le=19)] @@ -172,7 +173,7 @@ def dump_to_primitive_dict(self): class Database(BaseModel): type: DatabaseEngine = DatabaseEngine.MARIADB - host: Host = "localhost" + host: DomainStr = "localhost" port: Port = 3306 name: NonEmptyStr = "clp-db" ssl_cert: Optional[NonEmptyStr] = None @@ -261,7 +262,7 @@ class CompressionScheduler(BaseModel): class QueryScheduler(BaseModel): - host: Host = "localhost" + host: DomainStr = "localhost" port: Port = 7000 jobs_poll_delay: PositiveFloat = 0.1 # seconds num_archives_to_search_per_sub_job: PositiveInt = 16 @@ -277,7 +278,7 @@ class QueryWorker(BaseModel): class Redis(BaseModel): - host: Host = "localhost" + host: DomainStr = "localhost" port: Port = 6379 query_backend_database: int = 0 compression_backend_database: int = 1 @@ -306,14 +307,14 @@ def load_credentials_from_env(self): class Reducer(BaseModel): - host: Host = "localhost" + host: DomainStr = "localhost" base_port: Port = 14009 logging_level: LoggingLevel = "INFO" upsert_interval: PositiveInt = 100 # milliseconds class ResultsCache(BaseModel): - host: Host = "localhost" + host: DomainStr = "localhost" port: Port = 27017 db_name: NonEmptyStr = "clp-query-results" stream_collection_name: NonEmptyStr = "stream-files" @@ -324,7 +325,7 @@ def get_uri(self): class Queue(BaseModel): - host: Host = "localhost" + host: DomainStr = "localhost" port: Port = 5672 username: Optional[NonEmptyStr] = None @@ -543,7 +544,7 @@ def dump_to_primitive_dict(self): class WebUi(BaseModel): - host: Host = "localhost" + host: DomainStr = "localhost" port: Port = 4000 results_metadata_collection_name: NonEmptyStr = "results-metadata" rate_limit: PositiveInt = 1000 @@ -562,7 +563,7 @@ class GarbageCollector(BaseModel): class Presto(BaseModel): - host: Host + host: DomainStr port: Port From 04929ea358620e0143a3747e9ac6a0d474096250 Mon Sep 17 00:00:00 2001 From: sitaowang1998 Date: Fri, 10 Oct 2025 11:31:26 -0400 Subject: [PATCH 16/33] Add custom annotation for serialization --- .../clp-py-utils/clp_py_utils/clp_config.py | 95 ++++++------------- .../pydantic_serialization_utils.py | 8 ++ 2 files changed, 37 insertions(+), 66 deletions(-) create mode 100644 components/clp-py-utils/clp_py_utils/pydantic_serialization_utils.py diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index 4d408dcc75..020c28ca30 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -21,6 +21,7 @@ read_yaml_config_file, validate_path_could_be_dir, ) +from .pydantic_serialization_utils import PathStr, StrEnumSerializer # Constants # Component names @@ -114,17 +115,26 @@ class StorageEngine(KebabCaseStrEnum): CLP_S = auto() +StorageEngineStr = Annotated[StorageEngine, StrEnumSerializer] + + class DatabaseEngine(KebabCaseStrEnum): MARIADB = auto() MYSQL = auto() +DatabaseEngineStr = Annotated[DatabaseEngine, StrEnumSerializer] + + class QueryEngine(KebabCaseStrEnum): CLP = auto() CLP_S = auto() PRESTO = auto() +QueryEngineStr = Annotated[QueryEngine, StrEnumSerializer] + + class StorageType(LowercaseStrEnum): FS = auto() S3 = auto() @@ -137,9 +147,12 @@ class AwsAuthType(LowercaseStrEnum): ec2 = auto() +AwsAuthTypeStr = Annotated[AwsAuthType, StrEnumSerializer] + + class Package(BaseModel): - storage_engine: StorageEngine = StorageEngine.CLP - query_engine: QueryEngine = QueryEngine.CLP + storage_engine: StorageEngineStr = StorageEngine.CLP + query_engine: QueryEngineStr = QueryEngine.CLP @model_validator(mode="after") def validate_query_engine_package_compatibility(self): @@ -163,15 +176,9 @@ def validate_query_engine_package_compatibility(self): return self - def dump_to_primitive_dict(self): - d = self.model_dump() - d["storage_engine"] = d["storage_engine"].value - d["query_engine"] = d["query_engine"].value - return d - class Database(BaseModel): - type: DatabaseEngine = DatabaseEngine.MARIADB + type: DatabaseEngineStr = DatabaseEngine.MARIADB host: DomainStr = "localhost" port: Port = 3306 name: NonEmptyStr = "clp-db" @@ -232,7 +239,6 @@ def get_clp_connection_params_and_type(self, disable_localhost_socket_connection def dump_to_primitive_dict(self): d = self.model_dump(exclude={"username", "password"}) - d["type"] = d["type"].value return d def load_credentials_from_file(self, credentials_file_path: pathlib.Path): @@ -360,12 +366,7 @@ class S3Credentials(BaseModel): class AwsAuthentication(BaseModel): - type: Literal[ - AwsAuthType.credentials.value, - AwsAuthType.profile.value, - AwsAuthType.env_vars.value, - AwsAuthType.ec2.value, - ] + type: AwsAuthTypeStr profile: Optional[NonEmptyStr] = None credentials: Optional[S3Credentials] = None @@ -408,13 +409,10 @@ class S3IngestionConfig(BaseModel): type: Literal[StorageType.S3.value] = StorageType.S3.value aws_authentication: AwsAuthentication - def dump_to_primitive_dict(self): - return self.model_dump() - class FsStorage(BaseModel): type: Literal[StorageType.FS.value] = StorageType.FS.value - directory: pathlib.Path + directory: PathStr @field_validator("directory", mode="before") @classmethod @@ -425,16 +423,11 @@ def validate_directory(cls, value): def make_config_paths_absolute(self, clp_home: pathlib.Path): self.directory = make_config_path_absolute(clp_home, self.directory) - def dump_to_primitive_dict(self): - d = self.model_dump() - d["directory"] = str(d["directory"]) - return d - class S3Storage(BaseModel): type: Literal[StorageType.S3.value] = StorageType.S3.value s3_config: S3Config - staging_directory: pathlib.Path + staging_directory: PathStr @field_validator("staging_directory", mode="before") @classmethod @@ -455,30 +448,25 @@ def validate_key_prefix(cls, value): def make_config_paths_absolute(self, clp_home: pathlib.Path): self.staging_directory = make_config_path_absolute(clp_home, self.staging_directory) - def dump_to_primitive_dict(self): - d = self.model_dump() - d["staging_directory"] = str(d["staging_directory"]) - return d - class FsIngestionConfig(FsStorage): - directory: pathlib.Path = pathlib.Path("/") + directory: PathStr = pathlib.Path("/") class ArchiveFsStorage(FsStorage): - directory: pathlib.Path = CLP_DEFAULT_DATA_DIRECTORY_PATH / "archives" + directory: PathStr = CLP_DEFAULT_DATA_DIRECTORY_PATH / "archives" class StreamFsStorage(FsStorage): - directory: pathlib.Path = CLP_DEFAULT_DATA_DIRECTORY_PATH / "streams" + directory: PathStr = CLP_DEFAULT_DATA_DIRECTORY_PATH / "streams" class ArchiveS3Storage(S3Storage): - staging_directory: pathlib.Path = CLP_DEFAULT_DATA_DIRECTORY_PATH / "staged-archives" + staging_directory: PathStr = CLP_DEFAULT_DATA_DIRECTORY_PATH / "staged-archives" class StreamS3Storage(S3Storage): - staging_directory: pathlib.Path = CLP_DEFAULT_DATA_DIRECTORY_PATH / "staged-streams" + staging_directory: PathStr = CLP_DEFAULT_DATA_DIRECTORY_PATH / "staged-streams" def _get_directory_from_storage_config( @@ -520,11 +508,6 @@ def set_directory(self, directory: pathlib.Path): def get_directory(self) -> pathlib.Path: return _get_directory_from_storage_config(self.storage) - def dump_to_primitive_dict(self): - d = self.model_dump() - d["storage"] = self.storage.dump_to_primitive_dict() - return d - class StreamOutput(BaseModel): storage: Union[StreamFsStorage, StreamS3Storage] = StreamFsStorage() @@ -536,11 +519,6 @@ def set_directory(self, directory: pathlib.Path): def get_directory(self) -> pathlib.Path: return _get_directory_from_storage_config(self.storage) - def dump_to_primitive_dict(self): - d = self.model_dump() - d["storage"] = self.storage.dump_to_primitive_dict() - return d - class WebUi(BaseModel): host: DomainStr = "localhost" @@ -590,20 +568,18 @@ class CLPConfig(BaseModel): query_worker: QueryWorker = QueryWorker() webui: WebUi = WebUi() garbage_collector: GarbageCollector = GarbageCollector() - credentials_file_path: pathlib.Path = CLP_DEFAULT_CREDENTIALS_FILE_PATH + credentials_file_path: PathStr = CLP_DEFAULT_CREDENTIALS_FILE_PATH presto: Optional[Presto] = None archive_output: ArchiveOutput = ArchiveOutput() stream_output: StreamOutput = StreamOutput() - data_directory: pathlib.Path = pathlib.Path("var") / "data" - logs_directory: pathlib.Path = pathlib.Path("var") / "log" + data_directory: PathStr = pathlib.Path("var") / "data" + logs_directory: PathStr = pathlib.Path("var") / "log" aws_config_directory: Optional[pathlib.Path] = None - _container_image_id_path: pathlib.Path = PrivateAttr( - default=CLP_PACKAGE_CONTAINER_IMAGE_ID_PATH - ) - _version_file_path: pathlib.Path = PrivateAttr(default=CLP_VERSION_FILE_PATH) + _container_image_id_path: PathStr = PrivateAttr(default=CLP_PACKAGE_CONTAINER_IMAGE_ID_PATH) + _version_file_path: PathStr = PrivateAttr(default=CLP_VERSION_FILE_PATH) @field_validator("aws_config_directory") @classmethod @@ -748,9 +724,6 @@ def dump_to_primitive_dict(self): d[key] = getattr(self, key).dump_to_primitive_dict() # Turn paths into primitive strings - d["credentials_file_path"] = str(self.credentials_file_path) - d["data_directory"] = str(self.data_directory) - d["logs_directory"] = str(self.logs_directory) if self.aws_config_directory is not None: d["aws_config_directory"] = str(self.aws_config_directory) else: @@ -778,16 +751,6 @@ class WorkerConfig(BaseModel): stream_output: StreamOutput = StreamOutput() stream_collection_name: str = ResultsCache().stream_collection_name - def dump_to_primitive_dict(self): - d = self.model_dump() - d["archive_output"] = self.archive_output.dump_to_primitive_dict() - - # Turn paths into primitive strings - d["data_directory"] = str(self.data_directory) - d["stream_output"] = self.stream_output.dump_to_primitive_dict() - - return d - def get_components_for_target(target: str) -> Set[str]: if target in TARGET_TO_COMPONENTS: diff --git a/components/clp-py-utils/clp_py_utils/pydantic_serialization_utils.py b/components/clp-py-utils/clp_py_utils/pydantic_serialization_utils.py new file mode 100644 index 0000000000..83141724d5 --- /dev/null +++ b/components/clp-py-utils/clp_py_utils/pydantic_serialization_utils.py @@ -0,0 +1,8 @@ +import pathlib +from typing import Annotated + +from pydantic import PlainSerializer + +StrEnumSerializer = PlainSerializer(lambda enum_value: enum_value.value) + +PathStr = Annotated[pathlib.Path, PlainSerializer(lambda path_value: str(path_value))] From 57b2789691d4ec3c3dc74bc5513843819381da40 Mon Sep 17 00:00:00 2001 From: sitaowang1998 Date: Fri, 10 Oct 2025 11:44:14 -0400 Subject: [PATCH 17/33] Remove field without custom serialization --- components/clp-py-utils/clp_py_utils/clp_config.py | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index 020c28ca30..dc19ccd140 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -517,7 +517,7 @@ def set_directory(self, directory: pathlib.Path): _set_directory_for_storage_config(self.storage, directory) def get_directory(self) -> pathlib.Path: - return _get_directory_from_storage_config(self.storage) + return _get_directory_from_storage_config(self.pathlib.Path) class WebUi(BaseModel): @@ -711,13 +711,9 @@ def get_runnable_components(self) -> Set[str]: def dump_to_primitive_dict(self): custom_serialized_fields = { - "package", "database", "queue", "redis", - "logs_input", - "archive_output", - "stream_output", } d = self.model_dump(exclude=custom_serialized_fields) for key in custom_serialized_fields: From 0b6f43c226098455ce88f261a757ab374bfb48fd Mon Sep 17 00:00:00 2001 From: sitaowang1998 Date: Fri, 10 Oct 2025 11:59:30 -0400 Subject: [PATCH 18/33] Bug fix --- components/clp-py-utils/clp_py_utils/clp_config.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index dc19ccd140..e80940148d 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -517,7 +517,7 @@ def set_directory(self, directory: pathlib.Path): _set_directory_for_storage_config(self.storage, directory) def get_directory(self) -> pathlib.Path: - return _get_directory_from_storage_config(self.pathlib.Path) + return _get_directory_from_storage_config(self.storage) class WebUi(BaseModel): From e487f82c7c1e9cc5caa840d8dc645c4cb9eadf3a Mon Sep 17 00:00:00 2001 From: sitaowang1998 Date: Tue, 14 Oct 2025 11:46:26 -0400 Subject: [PATCH 19/33] Fix enum --- components/clp-py-utils/clp_py_utils/clp_config.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index e80940148d..21f0f63233 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -669,7 +669,7 @@ def validate_aws_config_dir(self): auth_configs.append(self.stream_output.storage.s3_config.aws_authentication) for auth in auth_configs: - if AwsAuthType.profile.value == auth.type: + if AwsAuthType.profile == auth.type: profile_auth_used = True break From 4cac503755e4c024707c5943e192efdcce53880e Mon Sep 17 00:00:00 2001 From: sitaowang1998 Date: Sat, 18 Oct 2025 00:54:29 -0400 Subject: [PATCH 20/33] Rename file and restructure --- .../clp-py-utils/clp_py_utils/clp_config.py | 7 +++++- .../pydantic_serialization_utils.py | 8 ------- .../clp_py_utils/serialization_utils.py | 22 +++++++++++++++++++ 3 files changed, 28 insertions(+), 9 deletions(-) delete mode 100644 components/clp-py-utils/clp_py_utils/pydantic_serialization_utils.py create mode 100644 components/clp-py-utils/clp_py_utils/serialization_utils.py diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index 21f0f63233..720030a5f9 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -10,6 +10,7 @@ Field, field_validator, model_validator, + PlainSerializer, PrivateAttr, ) from strenum import KebabCaseStrEnum, LowercaseStrEnum @@ -21,7 +22,7 @@ read_yaml_config_file, validate_path_could_be_dir, ) -from .pydantic_serialization_utils import PathStr, StrEnumSerializer +from .serialization_utils import serialize_enum, serialize_path # Constants # Component names @@ -109,6 +110,10 @@ Port = Annotated[int, Field(gt=0, lt=2**16)] ZstdCompressionLevel = Annotated[int, Field(ge=1, le=19)] +StrEnumSerializer = PlainSerializer(serialize_enum) + +PathStr = Annotated[pathlib.Path, PlainSerializer(serialize_path)] + class StorageEngine(KebabCaseStrEnum): CLP = auto() diff --git a/components/clp-py-utils/clp_py_utils/pydantic_serialization_utils.py b/components/clp-py-utils/clp_py_utils/pydantic_serialization_utils.py deleted file mode 100644 index 83141724d5..0000000000 --- a/components/clp-py-utils/clp_py_utils/pydantic_serialization_utils.py +++ /dev/null @@ -1,8 +0,0 @@ -import pathlib -from typing import Annotated - -from pydantic import PlainSerializer - -StrEnumSerializer = PlainSerializer(lambda enum_value: enum_value.value) - -PathStr = Annotated[pathlib.Path, PlainSerializer(lambda path_value: str(path_value))] diff --git a/components/clp-py-utils/clp_py_utils/serialization_utils.py b/components/clp-py-utils/clp_py_utils/serialization_utils.py new file mode 100644 index 0000000000..4d0526bcce --- /dev/null +++ b/components/clp-py-utils/clp_py_utils/serialization_utils.py @@ -0,0 +1,22 @@ +import pathlib +from enum import StrEnum + + +def serialize_enum(enum_value: StrEnum) -> str: + """ + Serializes a StrEnum to its underlying value. + + :param enum_value: A StrEnum instance. + :return: The underlying string value of the StrEnum. + """ + return enum_value.value() + + +def serialize_path(path: pathlib.Path) -> str: + """ + Serializes a pathlib.Path to its string representation. + + :param path: A pathlib.Path instance. + :return: The string representation of the path. + """ + return str(path) From da5e3e72afd1a14798c47281ad4633872df0bbef Mon Sep 17 00:00:00 2001 From: sitaowang1998 Date: Sat, 18 Oct 2025 01:00:16 -0400 Subject: [PATCH 21/33] Use before validator --- .../clp-py-utils/clp_py_utils/clp_config.py | 55 +++++++++---------- 1 file changed, 26 insertions(+), 29 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index 720030a5f9..2d375e737f 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -6,6 +6,7 @@ from dotenv import dotenv_values from pydantic import ( BaseModel, + BeforeValidator, ConfigDict, Field, field_validator, @@ -115,6 +116,29 @@ PathStr = Annotated[pathlib.Path, PlainSerializer(serialize_path)] +def _validate_directory(value: Any): + """ + Validates that the given value represents a directory path. + + :param value: + :return: `value` + :raise ValueError: if the value is not of type str. + :raise ValueError: if the value is an empty string. + """ + if not isinstance(value, str): + raise ValueError("must be a string.") + + if "" == value.strip(): + raise ValueError("cannot be empty") + + return value + + +ValidatedPathStr = Annotated[ + pathlib.Path, BeforeValidator(_validate_directory), PlainSerializer(serialize_path) +] + + class StorageEngine(KebabCaseStrEnum): CLP = auto() CLP_S = auto() @@ -417,13 +441,7 @@ class S3IngestionConfig(BaseModel): class FsStorage(BaseModel): type: Literal[StorageType.FS.value] = StorageType.FS.value - directory: PathStr - - @field_validator("directory", mode="before") - @classmethod - def validate_directory(cls, value): - _validate_directory(value) - return value + directory: ValidatedPathStr def make_config_paths_absolute(self, clp_home: pathlib.Path): self.directory = make_config_path_absolute(clp_home, self.directory) @@ -432,13 +450,7 @@ def make_config_paths_absolute(self, clp_home: pathlib.Path): class S3Storage(BaseModel): type: Literal[StorageType.S3.value] = StorageType.S3.value s3_config: S3Config - staging_directory: PathStr - - @field_validator("staging_directory", mode="before") - @classmethod - def validate_staging_directory(cls, value): - _validate_directory(value) - return value + staging_directory: ValidatedPathStr @field_validator("s3_config") @classmethod @@ -760,18 +772,3 @@ def get_components_for_target(target: str) -> Set[str]: return {target} else: return set() - - -def _validate_directory(value: Any): - """ - Validates that the given value represents a directory path. - - :param value: - :raise ValueError: if the value is not of type str. - :raise ValueError: if the value is an empty string. - """ - if not isinstance(value, str): - raise ValueError("must be a string.") - - if "" == value.strip(): - raise ValueError("cannot be empty") From bfa44c6fce5fbaaa2a80eaddc4a68c6afadbc7c6 Mon Sep 17 00:00:00 2001 From: sitaowang1998 Date: Sat, 18 Oct 2025 01:26:35 -0400 Subject: [PATCH 22/33] Bug fix --- components/clp-py-utils/clp_py_utils/serialization_utils.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/components/clp-py-utils/clp_py_utils/serialization_utils.py b/components/clp-py-utils/clp_py_utils/serialization_utils.py index 4d0526bcce..7c574749a0 100644 --- a/components/clp-py-utils/clp_py_utils/serialization_utils.py +++ b/components/clp-py-utils/clp_py_utils/serialization_utils.py @@ -9,7 +9,7 @@ def serialize_enum(enum_value: StrEnum) -> str: :param enum_value: A StrEnum instance. :return: The underlying string value of the StrEnum. """ - return enum_value.value() + return enum_value.value def serialize_path(path: pathlib.Path) -> str: From f6abd555d613cd23b3b10ef26e78936f3677fc04 Mon Sep 17 00:00:00 2001 From: sitaowang1998 Date: Sat, 18 Oct 2025 01:32:53 -0400 Subject: [PATCH 23/33] Bug fix --- components/clp-py-utils/clp_py_utils/serialization_utils.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/components/clp-py-utils/clp_py_utils/serialization_utils.py b/components/clp-py-utils/clp_py_utils/serialization_utils.py index 7c574749a0..81446b5359 100644 --- a/components/clp-py-utils/clp_py_utils/serialization_utils.py +++ b/components/clp-py-utils/clp_py_utils/serialization_utils.py @@ -1,5 +1,5 @@ import pathlib -from enum import StrEnum +from strenum import StrEnum def serialize_enum(enum_value: StrEnum) -> str: From c80de480f7cb6eab79108816dbb2ca5d46631067 Mon Sep 17 00:00:00 2001 From: sitaowang1998 Date: Sat, 18 Oct 2025 01:54:01 -0400 Subject: [PATCH 24/33] Revert "Use before validator" This reverts commit da5e3e72afd1a14798c47281ad4633872df0bbef. --- .../clp-py-utils/clp_py_utils/clp_config.py | 55 ++++++++++--------- 1 file changed, 29 insertions(+), 26 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index 2d375e737f..720030a5f9 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -6,7 +6,6 @@ from dotenv import dotenv_values from pydantic import ( BaseModel, - BeforeValidator, ConfigDict, Field, field_validator, @@ -116,29 +115,6 @@ PathStr = Annotated[pathlib.Path, PlainSerializer(serialize_path)] -def _validate_directory(value: Any): - """ - Validates that the given value represents a directory path. - - :param value: - :return: `value` - :raise ValueError: if the value is not of type str. - :raise ValueError: if the value is an empty string. - """ - if not isinstance(value, str): - raise ValueError("must be a string.") - - if "" == value.strip(): - raise ValueError("cannot be empty") - - return value - - -ValidatedPathStr = Annotated[ - pathlib.Path, BeforeValidator(_validate_directory), PlainSerializer(serialize_path) -] - - class StorageEngine(KebabCaseStrEnum): CLP = auto() CLP_S = auto() @@ -441,7 +417,13 @@ class S3IngestionConfig(BaseModel): class FsStorage(BaseModel): type: Literal[StorageType.FS.value] = StorageType.FS.value - directory: ValidatedPathStr + directory: PathStr + + @field_validator("directory", mode="before") + @classmethod + def validate_directory(cls, value): + _validate_directory(value) + return value def make_config_paths_absolute(self, clp_home: pathlib.Path): self.directory = make_config_path_absolute(clp_home, self.directory) @@ -450,7 +432,13 @@ def make_config_paths_absolute(self, clp_home: pathlib.Path): class S3Storage(BaseModel): type: Literal[StorageType.S3.value] = StorageType.S3.value s3_config: S3Config - staging_directory: ValidatedPathStr + staging_directory: PathStr + + @field_validator("staging_directory", mode="before") + @classmethod + def validate_staging_directory(cls, value): + _validate_directory(value) + return value @field_validator("s3_config") @classmethod @@ -772,3 +760,18 @@ def get_components_for_target(target: str) -> Set[str]: return {target} else: return set() + + +def _validate_directory(value: Any): + """ + Validates that the given value represents a directory path. + + :param value: + :raise ValueError: if the value is not of type str. + :raise ValueError: if the value is an empty string. + """ + if not isinstance(value, str): + raise ValueError("must be a string.") + + if "" == value.strip(): + raise ValueError("cannot be empty") From 1f9bfc95603594e5bba75904a01aa38a40cf50e0 Mon Sep 17 00:00:00 2001 From: sitaowang1998 Date: Sat, 18 Oct 2025 02:28:32 -0400 Subject: [PATCH 25/33] Fix lint --- components/clp-py-utils/clp_py_utils/serialization_utils.py | 1 + 1 file changed, 1 insertion(+) diff --git a/components/clp-py-utils/clp_py_utils/serialization_utils.py b/components/clp-py-utils/clp_py_utils/serialization_utils.py index 81446b5359..cb423f2baa 100644 --- a/components/clp-py-utils/clp_py_utils/serialization_utils.py +++ b/components/clp-py-utils/clp_py_utils/serialization_utils.py @@ -1,4 +1,5 @@ import pathlib + from strenum import StrEnum From ca09544ae956c0d2db387c5dd3939a6711b0d8a6 Mon Sep 17 00:00:00 2001 From: sitaowang1998 Date: Sat, 18 Oct 2025 11:15:33 -0400 Subject: [PATCH 26/33] Apply suggestions from code review Co-authored-by: Junhao Liao --- components/clp-py-utils/clp_py_utils/clp_config.py | 2 +- .../clp-py-utils/clp_py_utils/serialization_utils.py | 10 +++++----- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index 720030a5f9..62a79491a3 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -581,7 +581,7 @@ class CLPConfig(BaseModel): stream_output: StreamOutput = StreamOutput() data_directory: PathStr = pathlib.Path("var") / "data" logs_directory: PathStr = pathlib.Path("var") / "log" - aws_config_directory: Optional[pathlib.Path] = None + aws_config_directory: Optional[SerializablePath] = None _container_image_id_path: PathStr = PrivateAttr(default=CLP_PACKAGE_CONTAINER_IMAGE_ID_PATH) _version_file_path: PathStr = PrivateAttr(default=CLP_VERSION_FILE_PATH) diff --git a/components/clp-py-utils/clp_py_utils/serialization_utils.py b/components/clp-py-utils/clp_py_utils/serialization_utils.py index cb423f2baa..581bf4b00b 100644 --- a/components/clp-py-utils/clp_py_utils/serialization_utils.py +++ b/components/clp-py-utils/clp_py_utils/serialization_utils.py @@ -5,19 +5,19 @@ def serialize_enum(enum_value: StrEnum) -> str: """ - Serializes a StrEnum to its underlying value. + Serializes a `strenum.StrEnum` member to its underlying value. - :param enum_value: A StrEnum instance. - :return: The underlying string value of the StrEnum. + :param member: + :return: The underlying string value of the enum member. """ return enum_value.value def serialize_path(path: pathlib.Path) -> str: """ - Serializes a pathlib.Path to its string representation. + Serializes a `pathlib.Path` to its string representation. - :param path: A pathlib.Path instance. + :param path: :return: The string representation of the path. """ return str(path) From 60da51042a4cd0ac498d55e2f9e3f42ed3215e99 Mon Sep 17 00:00:00 2001 From: sitaowang1998 Date: Sat, 18 Oct 2025 11:18:40 -0400 Subject: [PATCH 27/33] Rename variables --- .../clp-py-utils/clp_py_utils/clp_config.py | 31 +++++++++---------- 1 file changed, 15 insertions(+), 16 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index 720030a5f9..27ac9044fe 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -100,6 +100,8 @@ CLP_QUEUE_PASS_ENV_VAR_NAME = "CLP_QUEUE_PASS" CLP_REDIS_PASS_ENV_VAR_NAME = "CLP_REDIS_PASS" +#Serializer +StrEnumSerializer = PlainSerializer(serialize_enum) # Generic types NonEmptyStr = Annotated[str, Field(min_length=1)] PositiveFloat = Annotated[float, Field(gt=0)] @@ -109,10 +111,7 @@ DomainStr = NonEmptyStr Port = Annotated[int, Field(gt=0, lt=2**16)] ZstdCompressionLevel = Annotated[int, Field(ge=1, le=19)] - -StrEnumSerializer = PlainSerializer(serialize_enum) - -PathStr = Annotated[pathlib.Path, PlainSerializer(serialize_path)] +SerializablePath = Annotated[pathlib.Path, PlainSerializer(serialize_path)] class StorageEngine(KebabCaseStrEnum): @@ -417,7 +416,7 @@ class S3IngestionConfig(BaseModel): class FsStorage(BaseModel): type: Literal[StorageType.FS.value] = StorageType.FS.value - directory: PathStr + directory: SerializablePath @field_validator("directory", mode="before") @classmethod @@ -432,7 +431,7 @@ def make_config_paths_absolute(self, clp_home: pathlib.Path): class S3Storage(BaseModel): type: Literal[StorageType.S3.value] = StorageType.S3.value s3_config: S3Config - staging_directory: PathStr + staging_directory: SerializablePath @field_validator("staging_directory", mode="before") @classmethod @@ -455,23 +454,23 @@ def make_config_paths_absolute(self, clp_home: pathlib.Path): class FsIngestionConfig(FsStorage): - directory: PathStr = pathlib.Path("/") + directory: SerializablePath = pathlib.Path("/") class ArchiveFsStorage(FsStorage): - directory: PathStr = CLP_DEFAULT_DATA_DIRECTORY_PATH / "archives" + directory: SerializablePath = CLP_DEFAULT_DATA_DIRECTORY_PATH / "archives" class StreamFsStorage(FsStorage): - directory: PathStr = CLP_DEFAULT_DATA_DIRECTORY_PATH / "streams" + directory: SerializablePath = CLP_DEFAULT_DATA_DIRECTORY_PATH / "streams" class ArchiveS3Storage(S3Storage): - staging_directory: PathStr = CLP_DEFAULT_DATA_DIRECTORY_PATH / "staged-archives" + staging_directory: SerializablePath = CLP_DEFAULT_DATA_DIRECTORY_PATH / "staged-archives" class StreamS3Storage(S3Storage): - staging_directory: PathStr = CLP_DEFAULT_DATA_DIRECTORY_PATH / "staged-streams" + staging_directory: SerializablePath = CLP_DEFAULT_DATA_DIRECTORY_PATH / "staged-streams" def _get_directory_from_storage_config( @@ -573,18 +572,18 @@ class CLPConfig(BaseModel): query_worker: QueryWorker = QueryWorker() webui: WebUi = WebUi() garbage_collector: GarbageCollector = GarbageCollector() - credentials_file_path: PathStr = CLP_DEFAULT_CREDENTIALS_FILE_PATH + credentials_file_path: SerializablePath = CLP_DEFAULT_CREDENTIALS_FILE_PATH presto: Optional[Presto] = None archive_output: ArchiveOutput = ArchiveOutput() stream_output: StreamOutput = StreamOutput() - data_directory: PathStr = pathlib.Path("var") / "data" - logs_directory: PathStr = pathlib.Path("var") / "log" + data_directory: SerializablePath = pathlib.Path("var") / "data" + logs_directory: SerializablePath = pathlib.Path("var") / "log" aws_config_directory: Optional[pathlib.Path] = None - _container_image_id_path: PathStr = PrivateAttr(default=CLP_PACKAGE_CONTAINER_IMAGE_ID_PATH) - _version_file_path: PathStr = PrivateAttr(default=CLP_VERSION_FILE_PATH) + _container_image_id_path: SerializablePath = PrivateAttr(default=CLP_PACKAGE_CONTAINER_IMAGE_ID_PATH) + _version_file_path: SerializablePath = PrivateAttr(default=CLP_VERSION_FILE_PATH) @field_validator("aws_config_directory") @classmethod From 0a7b1570982ee61c74783a12ccf4c5d14093026d Mon Sep 17 00:00:00 2001 From: sitaowang1998 Date: Sat, 18 Oct 2025 11:20:44 -0400 Subject: [PATCH 28/33] Rename variable --- components/clp-py-utils/clp_py_utils/serialization_utils.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/serialization_utils.py b/components/clp-py-utils/clp_py_utils/serialization_utils.py index 581bf4b00b..e0cc951b8b 100644 --- a/components/clp-py-utils/clp_py_utils/serialization_utils.py +++ b/components/clp-py-utils/clp_py_utils/serialization_utils.py @@ -3,14 +3,14 @@ from strenum import StrEnum -def serialize_enum(enum_value: StrEnum) -> str: +def serialize_enum(member: StrEnum) -> str: """ Serializes a `strenum.StrEnum` member to its underlying value. :param member: :return: The underlying string value of the enum member. """ - return enum_value.value + return member.value def serialize_path(path: pathlib.Path) -> str: From 78c015bc90c64683fff2c62aa6f7a2c5d8a48ab5 Mon Sep 17 00:00:00 2001 From: sitaowang1998 Date: Sat, 18 Oct 2025 11:22:36 -0400 Subject: [PATCH 29/33] Rename function --- components/clp-py-utils/clp_py_utils/clp_config.py | 10 ++++++---- .../clp-py-utils/clp_py_utils/serialization_utils.py | 2 +- 2 files changed, 7 insertions(+), 5 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index 075c2ec642..8f723ca92c 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -22,7 +22,7 @@ read_yaml_config_file, validate_path_could_be_dir, ) -from .serialization_utils import serialize_enum, serialize_path +from .serialization_utils import serialize_path, serialize_str_enum # Constants # Component names @@ -100,8 +100,8 @@ CLP_QUEUE_PASS_ENV_VAR_NAME = "CLP_QUEUE_PASS" CLP_REDIS_PASS_ENV_VAR_NAME = "CLP_REDIS_PASS" -#Serializer -StrEnumSerializer = PlainSerializer(serialize_enum) +# Serializer +StrEnumSerializer = PlainSerializer(serialize_str_enum) # Generic types NonEmptyStr = Annotated[str, Field(min_length=1)] PositiveFloat = Annotated[float, Field(gt=0)] @@ -582,7 +582,9 @@ class CLPConfig(BaseModel): logs_directory: SerializablePath = pathlib.Path("var") / "log" aws_config_directory: Optional[SerializablePath] = None - _container_image_id_path: SerializablePath = PrivateAttr(default=CLP_PACKAGE_CONTAINER_IMAGE_ID_PATH) + _container_image_id_path: SerializablePath = PrivateAttr( + default=CLP_PACKAGE_CONTAINER_IMAGE_ID_PATH + ) _version_file_path: SerializablePath = PrivateAttr(default=CLP_VERSION_FILE_PATH) @field_validator("aws_config_directory") diff --git a/components/clp-py-utils/clp_py_utils/serialization_utils.py b/components/clp-py-utils/clp_py_utils/serialization_utils.py index e0cc951b8b..2ebd0e0e78 100644 --- a/components/clp-py-utils/clp_py_utils/serialization_utils.py +++ b/components/clp-py-utils/clp_py_utils/serialization_utils.py @@ -3,7 +3,7 @@ from strenum import StrEnum -def serialize_enum(member: StrEnum) -> str: +def serialize_str_enum(member: StrEnum) -> str: """ Serializes a `strenum.StrEnum` member to its underlying value. From 7b3e9d7209670afc8fcc7aa1ed42723c3f12cc6b Mon Sep 17 00:00:00 2001 From: sitaowang1998 Date: Sat, 18 Oct 2025 11:25:19 -0400 Subject: [PATCH 30/33] Use optional serializable path --- components/clp-py-utils/clp_py_utils/clp_config.py | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index 8f723ca92c..f4244e4f26 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -589,7 +589,7 @@ class CLPConfig(BaseModel): @field_validator("aws_config_directory") @classmethod - def expand_profile_user_home(cls, value: Optional[pathlib.Path]): + def expand_profile_user_home(cls, value: Optional[SerializablePath]): if value is not None: value = value.expanduser() return value @@ -725,12 +725,6 @@ def dump_to_primitive_dict(self): for key in custom_serialized_fields: d[key] = getattr(self, key).dump_to_primitive_dict() - # Turn paths into primitive strings - if self.aws_config_directory is not None: - d["aws_config_directory"] = str(self.aws_config_directory) - else: - d["aws_config_directory"] = None - return d @model_validator(mode="after") From d1a36d6ce2540656246d30aac48dea09b3b76f38 Mon Sep 17 00:00:00 2001 From: sitaowang1998 Date: Sat, 18 Oct 2025 11:40:03 -0400 Subject: [PATCH 31/33] Add return type --- components/clp-py-utils/clp_py_utils/clp_config.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index f4244e4f26..90f324a096 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -589,7 +589,9 @@ class CLPConfig(BaseModel): @field_validator("aws_config_directory") @classmethod - def expand_profile_user_home(cls, value: Optional[SerializablePath]): + def expand_profile_user_home( + cls, value: Optional[SerializablePath] + ) -> Optional[SerializablePath]: if value is not None: value = value.expanduser() return value From a6b6c8d9600a4d9e0f3ba2dd2e0af00324f11473 Mon Sep 17 00:00:00 2001 From: sitaowang1998 Date: Sat, 18 Oct 2025 13:53:06 -0400 Subject: [PATCH 32/33] Fix type definition order Co-authored-by: Junhao Liao --- components/clp-py-utils/clp_py_utils/clp_config.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index 90f324a096..51766fb9ef 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -110,8 +110,8 @@ # TODO: Replace this with pydantic_extra_types.domain.DomainStr. DomainStr = NonEmptyStr Port = Annotated[int, Field(gt=0, lt=2**16)] -ZstdCompressionLevel = Annotated[int, Field(ge=1, le=19)] SerializablePath = Annotated[pathlib.Path, PlainSerializer(serialize_path)] +ZstdCompressionLevel = Annotated[int, Field(ge=1, le=19)] class StorageEngine(KebabCaseStrEnum): From 80a7164830618df33db97e0556e8e13dddcb621f Mon Sep 17 00:00:00 2001 From: sitaowang1998 Date: Sat, 18 Oct 2025 13:55:21 -0400 Subject: [PATCH 33/33] Replace or pathlib.Path --- components/clp-py-utils/clp_py_utils/clp_config.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index 51766fb9ef..87e0edb82d 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -743,7 +743,7 @@ def validate_presto_config(self): class WorkerConfig(BaseModel): package: Package = Package() archive_output: ArchiveOutput = ArchiveOutput() - data_directory: pathlib.Path = CLPConfig().data_directory + data_directory: SerializablePath = CLPConfig().data_directory # Only needed by query workers. stream_output: StreamOutput = StreamOutput()