refactor: extract settings data types out of lib module

python/cocoindex/cli.py

@@ -1,10 +1,9 @@
 import click
 import datetime
-import urllib.parse
 
 from rich.console import Console
 
-from . import flow, lib
+from . import flow, lib, setting
 from .flow import flow_names
 from .setup import sync_setup, drop_setup, flow_names_with_setup, apply_setup_changes
 from .runtime import execution_context
@@ -60,9 +59,9 @@ def show(flow_name: str | None, color: bool):
     """
     Show the flow spec in a readable format with colored output.
     """
-    flow = _flow_by_name(flow_name)
+    fl = _flow_by_name(flow_name)
     console = Console(no_color=not color)
-    console.print(flow._render_text())
+    console.print(fl._render_text())
 
 @cli.command()
 def setup():
@@ -151,7 +150,7 @@ def evaluate(flow_name: str | None, output_dir: str | None, cache: bool = True):
     options = flow.EvaluateAndDumpOptions(output_dir=output_dir, use_cache=cache)
     fl.evaluate_and_dump(options)
 
-_default_server_settings = lib.ServerSettings.from_env()
+_default_server_settings = setting.ServerSettings.from_env()
 
 COCOINDEX_HOST = 'https://cocoindex.io'
 
@@ -191,7 +190,7 @@ def server(address: str, live_update: bool, quiet: bool, cors_origin: str | None
         cors_origins.add(COCOINDEX_HOST)
     if cors_local is not None:
         cors_origins.add(f"http://localhost:{cors_local}")
-    lib.start_server(lib.ServerSettings(address=address, cors_origins=list(cors_origins)))
+    lib.start_server(setting.ServerSettings(address=address, cors_origins=list(cors_origins)))
     if live_update:
         options = flow.FlowLiveUpdaterOptions(live_mode=True, print_stats=not quiet)
         execution_context.run(flow.update_all_flows(options))

python/cocoindex/lib.py

@@ -1,76 +1,23 @@
 """
 Library level functions and states.
 """
-import os
 import sys
 import functools
 import inspect
 
-from typing import Callable, Self, Any
-from dataclasses import dataclass
+from typing import Callable
 
 from . import _engine
-from . import flow, query, cli
+from . import flow, query, cli, setting
 from .convert import dump_engine_object
 
 
-def _load_field(target: dict[str, Any], name: str, env_name: str, required: bool = False,
-                parse: Callable[[str], Any] | None = None):
-    value = os.getenv(env_name)
-    if value is None:
-        if required:
-            raise ValueError(f"{env_name} is not set")
-    else:
-        target[name] = value if parse is None else parse(value)
-
-@dataclass
-class DatabaseConnectionSpec:
-    url: str
-    user: str | None = None
-    password: str | None = None
-
-@dataclass
-class Settings:
-    """Settings for the cocoindex library."""
-    database: DatabaseConnectionSpec
-
-    @classmethod
-    def from_env(cls) -> Self:
-        """Load settings from environment variables."""
-
-        db_kwargs: dict[str, str] = dict()
-        _load_field(db_kwargs, "url", "COCOINDEX_DATABASE_URL", required=True)
-        _load_field(db_kwargs, "user", "COCOINDEX_DATABASE_USER")
-        _load_field(db_kwargs, "password", "COCOINDEX_DATABASE_PASSWORD")
-        database = DatabaseConnectionSpec(**db_kwargs)
-        return cls(database=database)
-
-
-def init(settings: Settings):
+def init(settings: setting.Settings):
     """Initialize the cocoindex library."""
     _engine.init(dump_engine_object(settings))
 
-@dataclass
-class ServerSettings:
-    """Settings for the cocoindex server."""
-
-    # The address to bind the server to.
-    address: str = "127.0.0.1:8080"
-
-    # The origins of the clients (e.g. CocoInsight UI) to allow CORS from.
-    cors_origins: list[str] | None = None
-
-    @classmethod
-    def from_env(cls) -> Self:
-        """Load settings from environment variables."""
-        kwargs: dict[str, Any] = dict()
-        _load_field(kwargs, "address", "COCOINDEX_SERVER_ADDRESS")
-        _load_field(kwargs, "cors_origins", "COCOINDEX_SERVER_CORS_ORIGINS",
-                    parse=lambda s: [o for e in s.split(",") if (o := e.strip()) != ""])
-        return cls(**kwargs)
-
 
-def start_server(settings: ServerSettings):
+def start_server(settings: setting.ServerSettings):
     """Start the cocoindex server."""
     flow.ensure_all_flows_built()
     query.ensure_all_handlers_built()
@@ -81,7 +28,7 @@ def stop():
     _engine.stop()
 
 def main_fn(
-        settings: Settings | None = None,
+        settings: setting.Settings | None = None,
         cocoindex_cmd: str = 'cocoindex',
         ) -> Callable[[Callable], Callable]:
     """
@@ -92,7 +39,7 @@ def main_fn(
     """
 
     def _pre_init() -> None:
-        effective_settings = settings or Settings.from_env()
+        effective_settings = settings or setting.Settings.from_env()
         init(effective_settings)
 
     def _should_run_cli() -> bool:

python/cocoindex/setting.py

@@ -0,0 +1,62 @@
+"""
+Data types for settings of the cocoindex library.
+"""
+import os
+
+from typing import Callable, Self, Any
+from dataclasses import dataclass
+
+
+@dataclass
+class DatabaseConnectionSpec:
+    """
+    Connection spec for relational database.
+    Used by both internal and target storage.
+    """
+    url: str
+    user: str | None = None
+    password: str | None = None
+
+def _load_field(target: dict[str, Any], name: str, env_name: str, required: bool = False,
+                parse: Callable[[str], Any] | None = None):
+    value = os.getenv(env_name)
+    if value is None:
+        if required:
+            raise ValueError(f"{env_name} is not set")
+    else:
+        target[name] = value if parse is None else parse(value)
+
+@dataclass
+class Settings:
+    """Settings for the cocoindex library."""
+    database: DatabaseConnectionSpec
+
+    @classmethod
+    def from_env(cls) -> Self:
+        """Load settings from environment variables."""
+
+        db_kwargs: dict[str, str] = dict()
+        _load_field(db_kwargs, "url", "COCOINDEX_DATABASE_URL", required=True)
+        _load_field(db_kwargs, "user", "COCOINDEX_DATABASE_USER")
+        _load_field(db_kwargs, "password", "COCOINDEX_DATABASE_PASSWORD")
+        database = DatabaseConnectionSpec(**db_kwargs)
+        return cls(database=database)
+
+@dataclass
+class ServerSettings:
+    """Settings for the cocoindex server."""
+
+    # The address to bind the server to.
+    address: str = "127.0.0.1:8080"
+
+    # The origins of the clients (e.g. CocoInsight UI) to allow CORS from.
+    cors_origins: list[str] | None = None
+
+    @classmethod
+    def from_env(cls) -> Self:
+        """Load settings from environment variables."""
+        kwargs: dict[str, Any] = dict()
+        _load_field(kwargs, "address", "COCOINDEX_SERVER_ADDRESS")
+        _load_field(kwargs, "cors_origins", "COCOINDEX_SERVER_CORS_ORIGINS",
+                    parse=lambda s: [o for e in s.split(",") if (o := e.strip()) != ""])
+        return cls(**kwargs)