diff --git a/docs/branch/py4lexis4.md b/docs/branch/py4lexis4.md index 636785a..7677da3 100644 --- a/docs/branch/py4lexis4.md +++ b/docs/branch/py4lexis4.md @@ -20,9 +20,10 @@ Py4lexis 5.x replaces `python-irodsclient` with `irods_http_client`. This breaks | Cycle | Goal | Status | |-------|------|--------| -| 1 | Create iRODS HTTP fsspec client | 🟡 Planning | -| 2 | Update session management | ⚪ Planned | -| 3 | Integrate new repository | ⚪ Planned | +| 1 | Create iRODS HTTP fsspec client | ✅ Complete | +| 1a | Async optimization (httpx) | ✅ Complete | +| 2 | Session management & config refactor | 🟡 Planning | +| 3 | Repository integration (HTTP2IRODSLexisRepository) | ⚪ Planned | | 4 | Code cleanup analysis | ⚪ Planned | --- @@ -111,39 +112,65 @@ tests/owilix/core/fsspec/ --- -## Cycle 2: Session Management +## Cycle 2: Session Management & Config Refactor ✅ -**Goal**: Update OWILIXManager to work with py4lexis 5.x session handling. +**Goal**: Refactor `OWILIXManager` and `OWILIXConfig` into a clean package. Ensure robust session handling. -### Key Changes +**Status**: Complete + +### Implemented Changes + +1. **Package Restructure**: Created `owilix/core/manager/` package. +2. **Pydantic Configuration**: `OWILIXSettings`, `RepositoryConfig`, etc. +3. **Session Wrapper**: `OWILIXSession` with lazy-loaded iRODS. +4. **UI Integration**: Moved `ui.py` into manager package. +5. **Optional Dependencies**: Added `irods-tcp` extras group. +6. **Tests**: 12 unit tests for env, config, session, backward compat. + +### Final Package Structure -- Update `session` property to use new `LexisSession` API -- Handle token refresh via wrapper (like `OWIIrods` in test file) -- Maintain `.owi/.tokens/` compatibility -- When session management is working, create proper unit and integration tests -- When tests are working, change other tests to use this session object (we need a structured access to the remote servers) +``` +owilix/core/manager/ +├── __init__.py # Package exports +├── env.py # OWILIXEnv singleton +├── config.py # Pydantic config models +├── session.py # OWILIXSession wrapper +├── manager.py # OWILIXManager class +└── ui.py # Console and progress utilities +``` -### Files to Modify +### Backward Compatibility -- `owilix/core/manager.py` - Session property -- `owilix/core/lexis.py` - Lexis API wrappers +Old import paths still work via re-export shims: +- `owilix.core.env` → `owilix.core.manager.env` +- `owilix.core.ui` → `owilix.core.manager.ui` +- `owilix.core.manager` (file) → `owilix.core.manager` (package) --- ## Cycle 3: Repository Integration -**Goal**: Replace `LEXISIrodsHTTPRepository` with new HTTP-based implementation. +**Goal**: Create `HTTP2IRODSLexisRepository` - a new repository class. + +### Architecture + +- **Derive from** `FileBasedRepository` (not `IRODSRepository`). +- **Metadata**: Use `OWILexisDatasetAPI` (LEXIS DDI HTTP API). +- **Files**: Use `Http2IrodsFileSystem` (iRODS HTTP API via fsspec). +- **Keep** `LEXISIrodsHTTPRepository` as legacy (optional dependency). ### Key Changes -- Create `LEXISHttp2IrodsRepository` using new fsspec -- Update `AggregatedRepository` integration -- Deprecate old `IRODSOWI` usage +1. Clean separation: DDI API for metadata ↔ iRODS HTTP API for files. +2. No `python-irodsclient` dependency in new repository. +3. Refactor `repository.py` for cleaner inheritance. -### Files to Modify +### Files to Create/Modify -- `owilix/core/repository.py` - New repository class -- Config handling for `http2irods` protocol +| Action | File | Description | +|--------|------|-------------| +| NEW | `core/repository/http2irods.py` | `HTTP2IRODSLexisRepository` | +| MODIFY | `core/repository.py` | Refactor base classes | --- diff --git a/docs/source/core.md b/docs/source/core.md new file mode 100644 index 0000000..876cb71 --- /dev/null +++ b/docs/source/core.md @@ -0,0 +1,157 @@ +# OWILIX Core Manager Package + +The `owilix.core.manager` package provides centralized configuration, session management, logging, and UI utilities. + +## Package Structure + +``` +owilix/core/manager/ +├── __init__.py # Package exports +├── env.py # OWILIXEnv singleton +├── config.py # Pydantic configuration models +├── session.py # OWILIXSession wrapper +├── manager.py # OWILIXManager class +├── ui.py # Console and progress utilities +└── logging.py # DBLog transaction logging +``` + +## Quick Start + +```python +from owilix.core.manager import OWILIXManager, OWILIXEnv + +# Environment configuration (singleton) +env = OWILIXEnv.values +print(f"Project: {env.project_name}") +print(f"Path: {env.owi_path}") + +# Manager (orchestrates config, session, repos) +manager = OWILIXManager() +session = manager.session # OWILIXSession wrapper +``` + +## Components + +### OWILIXEnv + +Singleton for environment variables. + +| Variable | Default | Description | +|----------|---------|-------------| +| `OWILIX_LEXIS_PROJECT_NAME` | `openwebsearch` | Project name | +| `OWILIX_LEXIS_PROJECT_ID` | Auto-generated | Project ID | +| `OWS_OWI_PATH` | `~/.owi` | Base path | +| `OWILIX_IRODS_CONNECTION_TIMEOUT` | `120` | Timeout (seconds) | + +```{eval-rst} +.. autoclass:: owilix.core.manager.OWILIXEnv + :members: +``` + +### OWILIXSettings (Pydantic) + +Type-safe configuration with validation. + +```python +from owilix.core.manager import OWILIXSettings, RepositoryConfig + +settings = OWILIXSettings() +repo = RepositoryConfig(repository="lexis+http", options={"zone": "IT4ILexisV2"}) +``` + +```{eval-rst} +.. autoclass:: owilix.core.manager.OWILIXSettings + :members: +``` + +### OWILIXSession + +Unified wrapper for `LexisSession` and iRODS access. + +```python +session = manager.session + +# Access LEXIS API +token = session.lexis.get_access_token() + +# Access iRODS (lazy-loaded) +irods = session.irods + +# Get HTTP client with fresh token +client = session.get_irods_client() +``` + +```{eval-rst} +.. autoclass:: owilix.core.manager.OWILIXSession + :members: +``` + +### OWILIXManager + +Central orchestrator for OWILIX operations. + +```{eval-rst} +.. autoclass:: owilix.core.manager.OWILIXManager + :members: + :show-inheritance: +``` + +## Logging + +### DBLog + +Thread-safe transaction logging for file processing with resume support. + +```python +from owilix.core.manager.logging import DBLog + +# Initialize for a job +dblog = DBLog("my_job", "/path/to/logs") + +# Register files +dblog.register_files([("/path/file1.parquet", "/dataset/path", "ds1")]) + +# Track progress +dblog.mark_files_success(["/path/file1.parquet"]) + +# Get statistics +stats = dblog.get_statistics() +``` + +```{eval-rst} +.. autoclass:: owilix.core.manager.logging.DBLog + :members: +``` + +## UI Utilities + +### OWILIXConsole + +Rich console with logging integration. + +```python +from owilix.core.manager import OWILIXConsole + +console = OWILIXConsole() +console.info("Processing started") +console.error("Something went wrong") +``` + +### EnhancedProgressDisplay + +Progress bars with error tracking. + +```python +from owilix.core.manager import EnhancedProgressDisplay, ErrorCollector + +errors = ErrorCollector() +progress = EnhancedProgressDisplay(console, errors) +progress.start(total_files=100) +progress.update(advance=1, records=1000) +progress.stop() +``` + +## See Also + +- [FSSPEC Integration](fsspec_integration.md) + diff --git a/owilix/cmd/base.py b/owilix/cmd/base.py index 03b316e..eefb3fd 100644 --- a/owilix/cmd/base.py +++ b/owilix/cmd/base.py @@ -27,7 +27,7 @@ from urllib.parse import urlparse from owilix.core.metadata import MetadataField from owilix.core.utils import dict_multi_key_get -from owilix.core.ui import * +from owilix.core.manager.ui import * from owilix.core.metadata import Dataset diff --git a/owilix/cmd/query.py b/owilix/cmd/query.py index 5d2f648..fd60c7d 100644 --- a/owilix/cmd/query.py +++ b/owilix/cmd/query.py @@ -59,7 +59,7 @@ Dependencies: """ from owilix.cmd.base import SubCommand, CommandResult, SQLBaseCommands -from owilix.core.ui import * +from owilix.core.manager.ui import * class QueryCommands(SQLBaseCommands): """ diff --git a/owilix/cmd/subcmds/query_extended.py b/owilix/cmd/subcmds/query_extended.py index 94c06cd..10c01f8 100644 --- a/owilix/cmd/subcmds/query_extended.py +++ b/owilix/cmd/subcmds/query_extended.py @@ -15,7 +15,7 @@ from owilix.core.ciff import drop_parallel from owilix.core.duckdb import OWIlixSQLQuery, OWIDuckDBCopyExecutor, OWIDuckDBSelectExecutor, ParquetBatchResult from owilix.core.metadata import Dataset, infer_metadata_from_files from owilix.core.stream import OWIDuckDBArrow -from owilix.core.ui import ErrorCollector, EnhancedProgressDisplay +from owilix.core.manager.ui import ErrorCollector, EnhancedProgressDisplay def aggregate(self, local_specifier: str, remote_specifier: str, diff --git a/owilix/core/__init__.py b/owilix/core/__init__.py index 8e725b6..80c3e02 100644 --- a/owilix/core/__init__.py +++ b/owilix/core/__init__.py @@ -1,4 +1,3 @@ -from .manager import OWIlixManager, load_and_check -from .env import _targetProjectHash -from .repository import LocalRepository, LEXISIrodsRepository, AggregatedRepository +from owilix.core.manager import OWIlixManager, load_and_check, _targetProjectHash +from .repository import LocalRepository, LEXISIrodsRepository, AggregatedRepository from .metadata import infer_metadata_from_files \ No newline at end of file diff --git a/owilix/core/duckdb.py b/owilix/core/duckdb.py index 75f7489..4bef507 100644 --- a/owilix/core/duckdb.py +++ b/owilix/core/duckdb.py @@ -13,8 +13,25 @@ from uuid import uuid4 import duckdb import fsspec from fsspec import AbstractFileSystem -from irods.exception import NetworkException, ExceptionOpenIDAuthUrl -from irods.session import iRODSSession + +# Handle python-irodsclient version differences +try: + from irods.exception import NetworkException +except ImportError: + NetworkException = Exception # Fallback if not installed + +try: + from irods.exception import ExceptionOpenIDAuthUrl +except ImportError: + # ExceptionOpenIDAuthUrl doesn't exist in newer versions + class ExceptionOpenIDAuthUrl(Exception): + """Placeholder for older python-irodsclient API.""" + pass + +try: + from irods.session import iRODSSession +except ImportError: + iRODSSession = None # Not installed _logger = logging.getLogger("owilix") diff --git a/owilix/core/irodsfsspec.py b/owilix/core/fsspec/irodsfsspec.py similarity index 100% rename from owilix/core/irodsfsspec.py rename to owilix/core/fsspec/irodsfsspec.py diff --git a/owilix/core/irods.py b/owilix/core/irods.py index 4b97e99..4c28762 100644 --- a/owilix/core/irods.py +++ b/owilix/core/irods.py @@ -3,7 +3,7 @@ import os from py4lexis.core.lexis_irods import iRODS from py4lexis.session import LexisSession -from owilix.core.env import OWILIXEnv +from owilix.core.manager import OWILIXEnv ## IRODS Client for the project. Allows maybe faster and more direct operations, including accessing iRODS metadata diff --git a/owilix/core/manager.py b/owilix/core/manager.py deleted file mode 100644 index 5f068d8..0000000 --- a/owilix/core/manager.py +++ /dev/null @@ -1,817 +0,0 @@ -""" - -OWILiX core manager module. - -This module provides: -- Configuration management (OWIlixConfig) with local and remote composition, version patching, and validation hooks. -- Project/session/repository orchestration (OWIlixManager), including repository setup, stats, logging, and secure token handling. -- Utility helpers for log rotation, flattening/counting tabular values, and generic file loading. - -Configuration overview (top-level keys): -- general: - - version: The OWILiX version string (auto-patched on first load if missing). - - server-update-url: Optional URL to fetch and merge server-provided configuration. - - license: License version the user agreed to (set by license_agreed). -- repositories: - - config: Mapping of repository name -> { repository: , options: {...} }. - Supported repository types: - - lexis: iRODS-based access - - lexis+http: iRODS metadata with HTTP data access - - s3a: S3-backed file access - - file: Local filesystem access - - selected_remote: List of repository names selected as remotes. - - selected_local: List of repository names selected as locals (typically includes "local"). -- profiles: Named display profiles with fields such as showfields, nodisplay, sort, and colorder. -- theme: Color settings used by CLI/UX layers. -- user: Default user metadata (name, email, organisation). -- dataset: Defaults for dataset handling (e.g., metadata). -- server-config: Optional section merged from server-update-url at runtime. - -Environment and security: -- Project name and ID are obtained via OWILIXEnv, sourced from environment variables where applicable. -- Refresh/offline tokens (for authentication) are stored under owi_path/.tokens with restrictive permissions. -- Server configuration is fetched with HTTP(S) when enabled; failures are logged and do not break local operation. - -Thread-safety: -- Lexis sessions are created behind a lock to avoid concurrent logins and ensure a single active session per process. - -Dependencies: -- Uses py4lexis for authentication/session handling and fsspec for file IO abstraction. -""" - -import hashlib -import json -import logging -import os -import re -import glob -import sys -import threading -from collections import defaultdict -from dataclasses import dataclass -from datetime import datetime -import requests -import stat -import fsspec -import pandas as pd -from owilix.core.validators import check_access -from py4lexis.core.exceptions import Py4LexisAuthException -from py4lexis.session import LexisSession -import yaml -from typing import Dict, Any - -from owilix.core.env import OWILIXEnv -from owilix.core.repository import LEXISIrodsRepository, LocalRepository, AggregatedRepository, \ - LEXISIrodsHTTPRepository, FileBasedRepository, S3FileBasedRepository -import os -import shutil -import getpass -from datetime import datetime -from owilix._version import __version__ -import dateutil - -from owilix.core.utils import dict_multi_key_add, dict_multi_key_get - - - - - -def rotate_log_file(filename, max_size = 500 * 1024 ): - - if os.path.isfile(filename): - file_size = os.path.getsize(filename) - if file_size > max_size: - # Create a new name for the rotated log file - timestamp = datetime.now().strftime('%Y%m%d%H%M%S') - rotated_filename = f"{filename}.{timestamp}" - shutil.copy2(filename, rotated_filename) # Copy the current log file to the new rotated log file - open(filename, 'w').close() # Truncate the original log file to create a new empty log file - - -def _flatten_and_count(df, column): - # flatten a column [a,b,c] and count the occurences of values. - if df[column].apply(lambda x: isinstance(x, list)).any(): - if df[column].explode().apply(lambda x: isinstance(x, list)).any(): - return df[column].explode().explode().value_counts().reset_index() # coudl be made generic, but no need yet - return df[column].explode().value_counts().reset_index() - else: - return df[column].value_counts().reset_index() - -# Example usage - -class OWIlixConfig: - """ - Represents the configuration handler for OWIlix, managing loading, saving, and manipulating - YAML-based configurations. This class handles local configurations, integrates remote - server configurations, validates schema conformity, and provides support for dynamic - repository and profile management. - - Attributes: - config_file (str): The path to the configuration file. - schema (Dict[str, Any] or None): The schema used for validation of the configuration, if available. - config (Dict[str, Any]): The current configuration loaded into memory. - - Methods: - __init__(config_file: str, schema: Dict[str, Any]=None, no_remote: bool=False): - Initializes the configuration handler with provided local configuration file, - optional schema, and controls remote server integration. - - delete_config(): - Deletes the existing configuration file and reloads default configuration. - - load_config(no_remote: bool=False) -> Dict[str, Any]: - Loads configuration from the YAML file, validates and patches it, and optionally - incorporates remote updates into the configuration. - - save_config() -> None: - Saves the current configuration back to the YAML file. - - validate_config(config: Dict[str, Any]) -> None: - Validates the specified configuration against the defined schema, if provided. - - patch_version(config: Dict[str, Any], target) -> Dict[str, Any]: - Applies necessary backward compatibility fixes for old configuration versions. - - add_config(key: str, value: Any) -> None: - Adds a new configuration entry using a multi-key format and validates the updated configuration. - - get_config(key: str) -> Any: - Retrieves a value from the configuration by key using multi-key lookup. - - get_repositories(remote: list=None, local: list=None): - Retrieves and filters repositories based on remote and local selection criteria, - allowing for backward compatibility handling. - - get_profile(name: str="default"): - Retrieves a profile configuration, combining defaults with specific overrides. - - get_theme(): - Retrieves the current theme configuration. - - license_agreed(license: Dict[str, Any]): - Updates the configuration to mark a specific license version as agreed. - - load_license(server_url: str="https://dashboard.ows.eu/api/license/latest") -> Dict[str, str]: - Fetches the license from a remote server and returns the details if different - from the current configuration. - - apply_server_config(local_config: Dict[str, Any], server_url: str="https://dashboard.ows.eu/api/owilix/default-cfg-0-17") -> Dict[str, Any]: - Applies configuration updates from a remote server to the local configuration. - """ - def __init__(self, config_file: str, - schema: Dict[str, Any]=None, - no_remote: bool=False): - self.config_file = config_file - self.schema = schema - self.config = self.load_config(no_remote) - - def delete_config(self): - if os.path.exists(self.config_file): - os.remove(self.config_file) - self.config_file= None - self.config = self.load_config() - - def _old_server_urls(self): # ugly, but no time to switch to proper version management. - return {"https://dashboard.ows.eu/api/owilix/default-cfg"} - - def load_config(self, no_remote:bool = False) -> Dict[str, Any]: - """Load the YAML configuration from file.""" - try: - with open(self.config_file, 'r') as file: - config = yaml.safe_load(file) # load config and patch it - _patched_config = self.patch_version(config, os.path.dirname(self.config_file)) - if not no_remote: - if "general" in _patched_config and "server-update-url" in _patched_config["general"]: # go here if there is a config entry for update-url - _url = _patched_config["general"].get ("server-update-url",None) - if _url in self._old_server_urls(): # use the current default urls and ignore old urls from this version onward. - _patched_config = self.apply_server_config(_patched_config) - else: - _patched_config = self.apply_server_config(_patched_config, _url) - else: - _patched_config = self.apply_server_config(_patched_config) - self.validate_config(_patched_config) - return _patched_config - except FileNotFoundError: - _default = self.get_default(os.path.dirname(self.config_file)) - if not no_remote: - _default = self.apply_server_config(_default) - return _default - - def save_config(self) -> None: - """Save the YAML configuration to file.""" - _dir_name = os.path.dirname(self.config_file) - if not os.path.exists(_dir_name): os.makedirs(_dir_name) - with open(self.config_file, 'w') as file: - yaml.dump(self.config, file) - - - def validate_config(self, config: Dict[str, Any]) -> None: - """Validate the configuration against the schema.""" - pass #validate(instance=config, schema=self.schema) - - def patch_version(self, config: Dict[str, Any], target) -> Dict[str, Any]: - """Validate the configuration against the schema.""" - if not "version" in config["general"]: # configuration before the time we stored the owilix version in the config - return self.get_default(target) - return config - - def add_config(self, key: str, value: Any) -> None: - """Add a new configuration entry using a multi key approach, i.e. key.key#positioin.key=value.""" - self.config = dict_multi_key_add(self.config, key, value) - self.validate_config(self.config) - - def get_config(self, key: str) -> Any: - """Get a configuration value by key.""" - return dict_multi_key_get(self.config, key) - - def get_repositories(self, remote:list=None, local:list=None): - """ - Retrieves and filters repositories based on provided remote and local selection criteria. - - If the remote list is not provided, the method selects all remotes specified in the - configuration by default. When the remote list contains entries prefixed with '+', these - are added to the default remotes from the configuration. Otherwise, the remote list overrides - the default selection entirely. - - For the local list, if it is not explicitly provided, the default selection is based on the - configuration. - - Additionally, the method applies a backward compatibility patch for older versions of the - configuration where 'name' was used instead of 'project_name' in the repository options. - - The method returns two dictionaries: one for the filtered remote repositories and another - for the filtered local repositories. - - Args: - remote (list, optional): A list of remote repository selectors. Defaults to None. - local (list, optional): A list of local repository selectors. Defaults to None. - - Returns: - tuple: A tuple containing two dictionaries: - - The first dictionary maps the keys of selected remote repositories to their configurations. - - The second dictionary maps the keys of selected local repositories to their configurations. - """ - repo_cfg =self.config.get("repositories",{}) - if remote is None: # we select all remotes if remote does not specifically select any remotes - remote= repo_cfg.get("selected_remote",[]) - elif all([r.startswith("+") for r in remote]): # add + in addition - remote = repo_cfg.get("selected_remote",[]) + [k[1:] for k in remote] - else: remote = [k[1:] for k in remote if k.startswith("+")] + [k for k in remote if not k.startswith("+")] - if local is None: - local = repo_cfg.get("selected_local",[]) - # do a patch here for old versions - for v in repo_cfg.get("config",{}).values(): - _options = v.get("options", {}) - if "name" in _options and "project_name" not in _options: - _options["project_name"] = _options.pop("name") - return {k:v for k,v in repo_cfg["config"].items() if k in remote}, \ - {k:v for k,v in repo_cfg["config"].items() if k in local} - - def get_profile(self, name="default"): - default = dict(self.config["profiles"]["default"]) - if name!="default": - default.update(self.config["profiles"].get(name, {})) - return default - - def get_theme(self): - return self.config.get("theme",{}) - - def license_agreed(self, license): - if "general" not in self.config: self.config["general"]={} - self.config["general"]["license"]=license["version"] - self.save_config() - - def load_license(self, server_url: str = "https://dashboard.ows.eu/api/license/latest") -> Dict[str, str]: - """ - Fetch the license from the server based on the specified version. - - This version first tests if the server is online with a HEAD request, - using a short timeout, so it doesn't delay the process if the server is offline. - - Args: - server_url (str): The URL to fetch the license from. - - Returns: - Dict[str, str]: The license data if successfully fetched and different - from the current version, otherwise {} or None. - """ - # Get the current license version from config if available - current = self.config.get("general", {}).get("license") - - # Step 1: Quickly check connectivity using a HEAD request with a short timeout - try: - head_response = requests.head(server_url, timeout=3) - head_response.raise_for_status() # Raise an error for bad HTTP status codes - except requests.RequestException as e: - print(f"Warning: Server is offline or not responding in a timely manner: {e}") - return {} # Exit early if server is not reachable - - # Step 2: Proceed to get the full license data if server is online - try: - response = requests.get(server_url, timeout=10) - response.raise_for_status() - json_data = response.json() - - # Only return new license data if it is different from the current version - return json_data if json_data.get("version") != current else None - except requests.RequestException as e: - print(f"Error fetching license: {e}") - return {} - - - def apply_server_config(self, local_config: Dict[str, Any], server_url: str="https://dashboard.ows.eu/api/owilix/default-cfg-0-17") -> Dict[str, Any]: - """ - Load server configuration and apply it to the local configuration. - Args: - local_config (Dict[str, Any]): The original local configuration. - server_url (str): The URL to fetch the server configuration. - - Returns: - Dict[str, Any]: The updated configuration with server-config and repository updates. - """ - if server_url is None or server_url =="": return local_config - - try: - # Fetch the server configuration - response = requests.get(server_url) - response.raise_for_status() - server_config = response.json() - - # Update repositories from the server config - if "repositories" in server_config and "config" in server_config["repositories"]: - for repo in server_config["repositories"]["config"].keys(): - local_config["repositories"]["config"][repo] = server_config["repositories"]["config"][repo] - - # Add or update the "server-config" section - if "server-config" in server_config: - local_config["server-config"] = server_config["server-config"] - - - return local_config - - except requests.RequestException as e: - logging.log(logging.ERROR, f"Error fetching repository configuration from server: {e}") - return local_config - - def get_default(self, target)-> Dict[str, Any]: - default_cfg = {} - default_cfg["repositories"]={} - default_cfg["general"]={"server-update-url":"https://dashboard.ows.eu/api/owilix/default-cfg-0-17", - "version": str(__version__)} - default_cfg["repositories"]["config"] = {"it4i": { - "repository": "lexis+http", - "options": { - "project_name": OWILIXEnv.values.project_name , - "lexis_id": OWILIXEnv.values.project_id, - "zone": "IT4ILexisV2" - }, - }, - "lrz": { - "repository": "lexis+http", - "options": { - "project_name": OWILIXEnv.values.project_name , - #'"host": "sikplrz-ows-icat.srv.mwn.de", - #"port": "1247", - "lexis_id": OWILIXEnv.values.project_id , - "zone": "IT4ILexisV2", - "zone_path": "OWSLRZZONE" # federation mode. If zone_path is different from zone, irods should federate - }, - }, - "local": { - "repository": "file", - "options": { - "path": target + os.path.sep + "{access}", - "collections_in_path": True - }, - }} - default_cfg["repositories"]["selected_remote"]=["it4i","lrz"] - default_cfg["repositories"]["selected_local"] = ["local"] - - default_cfg["profiles"]={ - "maxi": { - "showfields":"", - "nodisplay":"compression,encryption,contributor,path,metadataSource,"+ - "publicationYear,provenance,owner,publisher,creator,resourceTypeGeneral,[A-Z]+.*,containerIcon,containerParams,relatedSoftware," - "rights,rightsURI,workflowEx_id,workflow_id,month,day,year,type,relatedSoftware,license,publication", - "sort":"+startDate", - "colorder":"internalID,title,startDate,endDate,dataCenter,collectionName", - }, - "default": { - "showfields":"internalID,title,dataCenter,startDate,endDate,resourceType,access,collectionName,lastChanged,resourceType,rights,objectCount,totalSize,fileSize", - "nodisplay":".*", - "sort":"+startDate", - "colorder":"internalID,title,startDate,endDate,dataCenter,collectionName,resourceType,rights,objectCount,totalSize,fileSize", - }, - "mini": { - "showfields": "internalID,title,dataCenter", - "nodisplay": ".*", - "sort": "+startDate", - } - } - default_cfg["theme"]={ - "info": "dim cyan", - "warning": "magenta", - "danger": "bold red", - "error": "bold red" - } - default_cfg["user"]= { - "name" : getpass.getuser(), - "email": "", - "organisattion": "" - } - default_cfg["dataset"]= { - "metadata" : {} - } - return default_cfg - - - - - - - -class OWIlixManager: - """ - The OWIlixProject class is the central management point for the OpenWebSearch data through the `owilix` project. - - This class provides functionality for: - 1. Managing local configuration, including project name, logging, path, and log files. - 2. Accessing the LEXIS session for authorized data retrieval. - 3. Accessing datasets available locally and remotely. - - Attributes: - name (str): The project name, defaulting to "openwebsearch" or sourced from the `OWI_LEXIS_PROJECT_NAME` environment variable. - logger (logging.Logger): Logger instance to handle logging within the class. - owi_path (str): The path where the project resides. - logpath (str): The path where log files are stored. - logfile (str): The file path for logging events in JSON format. - _session (LexisSession or None): Private attribute for managing a LexisNexis session. - _remote_data (AggregatedRepository or None): Private attribute to hold an AggregatedRepository instance. - _local (Repository or None): Private attribute to hold an OWILocal instance. - - Properties: - local (OWILocal): Provides access to local data management functionalities. - remote_data (AggregatedRepository): Provides access to remote dataset functionalities. - session (LexisSession): Establishes or returns an existing LexisNexis session. - - Methods: - stats(day: str | datetime | None = None, duration: int = 0, data_center=None, query=None) -> pd.DataFrame: - Generates summary statistics for datasets within the specified constraints. - - update_log(data_dict: dict, slice: str | None = None) -> None: - Logs an event into a JSON file. - - load_log(slice: str | None = None) -> pd.DataFrame: - Loads log data from JSON files into a pandas DataFrame. - - parse_specifier(specifier: str) -> dict: - Parses a command specifier string to return filters and parameters for data queries. - - Example Usage: - ```python - # Create an instance of OWIlixProject - manager = OWILIXManager(owi_path='/path/to/manager', logpath='/path/to/logs') - - # Retrieve local data management instance - local_data = manager.local - - # Access remote datasets - remote_data = manager.datasets - - # Get a session with LexisNexis - session = manager.session - - # Update the log with some data - manager.update_log({'event': 'new_access', 'details': 'details here'}) - - # Load the log - log_df = manager.load_log() - - # Generate statistics for a specific day - stats_df = project.stats(day='2021-05-20') - """ - - def __init__(self, owi_path:str=OWILIXEnv.values.owi_path, - logpath:str=None, - repos:Dict[str,Dict]=None, - name:str="openwebsearch", - user:str|None = None, - config:OWIlixConfig=None, - no_refresh_token=False): - """ - Initializes the OWIlixProject instance with the specified paths. - The project name can be set via the `OWILIX_LEXIS_PROJECT_NAME` environment variable (overwriting the constructor parameter). - The project ID can be set via the `OWILIX_LEXIS_PROJECT_ID` environment variable (only needed for irods based access, oerwriting the constructor parameter). - - Args: - owi_path (str): Path to the project directory. - logpath (str): Path where log files will be stored. I None, it will be owi_path/.logs - repos (dict): Dictionary of repositories with their configurations. if None, repos will be instantiated from the config. - name (str): Project name, defaulting to "openwebsearch". - user (str): User name for the project. - config (OWIlixConfig): Configuration object for the project. If none, a standard config will be instantiated - no_refresh_token (bool): Does not use the refresh token, but creates a new one - """ - self.repos = repos - self.no_refresh_token = no_refresh_token - self.name = OWILIXEnv.values.project_name - self.lexis_id = OWILIXEnv.values.project_id - # configure logger. TODO: Set log level - self.logger = logging.getLogger("owilix") - handler = logging.StreamHandler(sys.stderr) # Defaults to stderr - formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') - handler.setFormatter(formatter) - self.logger.addHandler(handler) - - self.owi_path = owi_path - self.logpath = logpath or owi_path + os.path.sep + ".logs" - self._configpath = os.path.join(owi_path) - if not os.path.exists(self._configpath): os.makedirs(self._configpath) - self.config = config or OWIlixConfig(os.path.join(self._configpath, "owilix.cfg")) - - self._token_path = os.path.join(self._configpath, ".tokens") - if not os.path.exists(self._token_path): os.makedirs(self._token_path) - - if not repos: - remote, local = self.config.get_repositories() - repos = remote|local - if not os.path.exists(logpath): os.makedirs(logpath) - self.logfile = os.path.join(logpath,"events.json") - self._repo_stats_file = os.path.join(logpath,"repository_stats.json") - rotate_log_file(self.logfile) - _repos = {} - self._local = None - self._user = user - for name, config in repos.items(): - repo = None - if config["repository"] == "lexis": - repo = LEXISIrodsRepository(self, repo_name=name, **config["options"]) - elif config["repository"] == "lexis+http": - repo = LEXISIrodsHTTPRepository(self, repo_name=name, **config["options"]) - elif config["repository"] == "s3a": - repo = S3FileBasedRepository(self, repo_name=name, **config["options"]) - elif config["repository"] == "file": - repo = LocalRepository(self, repo_name=name, **config["options"]) - else: - raise ValueError(f"Repositorytype {config['repository']} unkwown.") - if name == "local": - self._local = repo - else: - _repos[name] = repo - self._remote_data= AggregatedRepository(_repos) - self._remote_data.load_stats_from_file(self._repo_stats_file) - if self.local == None: - raise ValueError("No local repository defined in the configuration") - self._session = None - self._user = None - self._session_lock = threading.Lock() - - - @property - def _refresh_token_fn(self): - return os.path.join(self._token_path, "refresh_token") - - - def clean(self, remove_cfg=False): - if os.path.exists(self._refresh_token_fn): - os.remove(self._refresh_token_fn) - if remove_cfg: - self.config.delete_config() - - @property - def local(self): - """ - Returns an instance of OWILocal for local data management, initializing it if not already created. - - Returns: - OWILocalData: Local data manager instance. - """ - return self._local - - @property - def user(self): - """ - Return user or none, if no user has been set. - """ - return self._user - - @property - def remote_data(self): - """ - Returns an instance of OWIRemoteData for accessing remote datasets, initializing it if not already created. - - Returns: - AggregatedRepository: Aggregated repository instance. - """ - return self._remote_data - - @property - def session(self): - """ - Returns an active LexisSession instance for authorized data retrieval, initializing it if not already created. - Thread-safe version. - - Returns: - LexisSession: LexisNexis session instance. - """ - if self._session is None: - with self._session_lock: - if self._session is None: # Double-checked locking - _login = self.config.get_config("login") - if _login.get("username", None) is not None: - self.logger.info("Using login/pwd from configuration") - self._session = LexisSession( - suppress_print=True, in_cli=True, - login_method="credentials", - log_file=self.get_lexis_log_filename(), - username=_login.get("username", None), - password=_login.get("password", None) - ) - return self._session - - _refresh_token = os.environ.get("PY4LEXIS_TOKEN", None) or self._refresh_token_io() - _refresh_token = None if self.no_refresh_token else _refresh_token - try: - if _refresh_token: - self._session = LexisSession( - suppress_print=True, in_cli=True, - offline_access=True, - login_method="offline", - log_file=self.get_lexis_log_filename(), - refresh_token=_refresh_token - ) - if self._session.get_access_token() is None: - raise Py4LexisAuthException("Token seems to be expired. Using URL login") - else: - raise Py4LexisAuthException("No refresh token found. Use URL login") - except Py4LexisAuthException: - self._session = LexisSession( - suppress_print=False, in_cli=True, - offline_access=True, - login_method="url", - log_file=self.get_lexis_log_filename() - ) - - _new_token = self._session.get_offline_token() or self._session.get_refresh_token() - if _new_token and _new_token != _refresh_token: - self._refresh_token_io(_new_token) - - return self._session - - def _refresh_token_io(self, token=None): - - if token is None: - try: - with open(self._refresh_token_fn, 'r') as file: - return file.read() - except FileNotFoundError: - return None - else: - with open(self._refresh_token_fn, 'w') as file: - file.write(token) - os.chmod(self._refresh_token_fn, stat.S_IRUSR | stat.S_IWUSR) - return token - - def stats(self, day: str|datetime|None=None, duration:int = 0, data_center=None, query=None): - """ - Generates statistics about datasets based on specified parameters. - - Args: - day (str | datetime | None): Specific day to analyze or 'None' for all available. - duration (int): Number of days to include from the specified day. - data_center (str | None): Data center name for filtering. - query (dict | None): Additional query parameters to filter datasets. - - Returns: - pd.DataFrame: Summary DataFrame with statistics on the queried datasets. - """ - # todo: extend by statistics for files - df, _files = self.remote_data.ls_http(day, duration, data_center, query) - # Initialize an empty DataFrame for the summary - summary_df = pd.DataFrame({'Kind':["Datasets"], "Key":["Total"], "Value":["Number Datasets"], "Count":[len(df)]}) - if df is None or len(df)==0: return summary_df - df['Month'] = pd.to_datetime(df['StartDate']).dt.to_period('M') - columns_to_exclude = ["Title", "InternalID", "CreationDate", "StartDate", "EndDate", "Path"] - df = df[[col for col in df.columns if col not in columns_to_exclude]] - # Loop through columns and append results - for column in df.columns: - temp_df = _flatten_and_count(df, column) - temp_df.columns = ['Value', 'Count'] - temp_df['Kind'] = "Datasets" - temp_df['Key'] = column - summary_df = pd.concat([summary_df, temp_df], ignore_index=True) - return summary_df - - - def update_log(self, data_dict): - """ - Appends an event to the log file in JSON format. - - Args: - data_dict (dict): Dictionary containing log data. - slice (str | None): Specific slice name for the log file, or None for the default file. - """ - data_dict["logdate"]=str(datetime.now().strftime('%Y-%m-%d %H:%M:%S')) - with open(self.logfile, "a") as file: - json_string = json.dumps(data_dict) - # Append the JSON string to the file with a newline - file.write(json_string + '\n') - - def load_log(self, all=True): - """ - Loads JSON-formatted log entries from a file into a pandas DataFrame. - - Returns: - list of dicts: list of dicts with all log entries - - """ - data = [] - log_files = glob.glob(os.path.join(self.logpath,"events*.json")) - for log_file in log_files: - with open(log_file, 'r') as file: - for line in file: - # Each line is a complete JSON object - if line.strip(): # Ensure that empty lines are skipped - json_block = json.loads(line.strip()) - data.append(json_block) - - # Convert list of dictionaries to DataFrame - return data - - - def parse_specifier(self, specifier): - """ - Parses a specifier string to extract data query parameters like data center, day, duration, and additional filters. - - Args: - specifier (str): Specifier string in the format `:#/=;=`. - - Returns: - dict: Parsed specifier with `data_center`, `day`, `duration`, and `query` keys. - """ - result = { - 'data_center': None, - 'day': None, - 'duration': None, - 'query': {} - } - - # Split based on '/' to separate main command and additional filters - parts = specifier.split('/') - main_part = parts[0] - filter_parts = parts[1:] if len(parts) > 1 else [] - # Regex to parse the main part of the command - main_regex = r'^(?P[\w-]+)(?::(?P[\d-]+|latest)(?:#(?P\d+))?)?$' - match = re.match(main_regex, main_part) - if match: - result.update(match.groupdict()) - - # If 'date' is specified, convert it to datetime object or handle 'latest' - if result['day']: - if result['day'] != 'latest': - try: - result['day'] = dateutil.parser.parse(result['day']) - except ValueError: - result['day'] = None - - # If 'days' is specified, convert it to integer - if result['duration']: - result['duration'] = int(result['duration']) - - if result.get('data_center', None) =="all": - result['data_center']=None - # Handle filters - for filter_part in filter_parts: - # Split filter_part on ';' to handle multiple key=value pairs - for kv in filter_part.split(';'): - if '=' in kv: - key, value = kv.split('=', 1) - result['query'][key] = value - - return result - - def get_lexis_log_filename(self): - return os.path.join(self.logpath, "lexis.log") - - def get_error_log_filename(self): - return os.path.join(self.logpath, "errors.log") - - def get_local_collection_path(self, access, collection): - check_access(access) - return os.path.join(self.owi_path, access, collection) - - def save_repository_stats(self): - self._remote_data.store_stats_to_file(self._repo_stats_file) - - - -def load_and_check(file, **kwargs): - """ Load a file as text using fsspec and check if it exists. - If the file exists, check whether it has formatting parameters like in **kwargs, but do not substitute them. - """ - # Open the file using fsspec - if len(file.split('://')) == 1: - file = 'file://' + file - with fsspec.open(file, 'r') as f: - return f.read() - diff --git a/owilix/core/manager/__init__.py b/owilix/core/manager/__init__.py new file mode 100644 index 0000000..fb7e93e --- /dev/null +++ b/owilix/core/manager/__init__.py @@ -0,0 +1,61 @@ +""" +OWILIX Core Manager Package + +This package provides centralized configuration and session management for OWILIX. + +Main components: +- OWILIXEnv: Environment configuration (singleton) +- OWILIXSettings: Pydantic configuration models +- OWILIXManager: Central orchestrator for repositories and sessions +- OWILIXSession: Unified session wrapper for LexisSession and iRODS +- OWILIXConsole: Rich console with logging integration +""" + +from owilix.core.manager.env import OWILIXEnv, _targetProjectHash +from owilix.core.manager.config import ( + OWILIXSettings, + RepositoryConfig, + ProfileConfig, +) +from owilix.core.manager.session import OWILIXSession +from owilix.core.manager.manager import OWILIXManager, OWIlixConfig, OWIlixManager +from owilix.core.manager.ui import ( + OWILIXConsole, + ErrorCategory, + QueryError, + ErrorCollector, + EnhancedProgressDisplay, +) + + +def load_and_check(file, **kwargs): + """Load a file as text using fsspec and check if it exists.""" + import fsspec + if len(file.split('://')) == 1: + file = 'file://' + file + with fsspec.open(file, 'r') as f: + return f.read() + + +__all__ = [ + # Environment + "OWILIXEnv", + "_targetProjectHash", + # Config + "OWILIXSettings", + "RepositoryConfig", + "ProfileConfig", + # Session + "OWILIXSession", + # Manager + "OWILIXManager", + "OWIlixConfig", + "OWIlixManager", + "load_and_check", + # UI + "OWILIXConsole", + "ErrorCategory", + "QueryError", + "ErrorCollector", + "EnhancedProgressDisplay", +] diff --git a/owilix/core/manager/config.py b/owilix/core/manager/config.py new file mode 100644 index 0000000..975f727 --- /dev/null +++ b/owilix/core/manager/config.py @@ -0,0 +1,186 @@ +""" +Pydantic Configuration Models for OWILIX + +This module provides type-safe configuration management using Pydantic. +Configurations can be loaded from YAML files or environment variables. +""" + +from typing import Dict, List, Literal, Optional, Any +from pydantic import BaseModel, Field, field_validator +from pydantic_settings import BaseSettings +import os + + +class RepositoryConfig(BaseModel): + """Configuration for a single repository. + + Attributes: + repository: Type of repository backend. + options: Backend-specific configuration options. + """ + repository: Literal["lexis", "lexis+http", "s3a", "file"] + options: Dict[str, Any] = Field(default_factory=dict) + + model_config = {"extra": "allow"} + + +class ProfileConfig(BaseModel): + """Display profile configuration. + + Attributes: + showfields: Comma-separated list of fields to display. + nodisplay: Regex pattern for fields to hide. + sort: Sort specification (e.g., "+startDate"). + colorder: Column ordering. + """ + showfields: str = "" + nodisplay: str = ".*" + sort: str = "+startDate" + colorder: str = "" + + model_config = {"extra": "allow"} + + +class ThemeConfig(BaseModel): + """CLI theme colors.""" + info: str = "dim cyan" + warning: str = "magenta" + danger: str = "bold red" + error: str = "bold red" + + model_config = {"extra": "allow"} + + +class UserConfig(BaseModel): + """User metadata.""" + name: str = "" + email: str = "" + organisation: str = "" + + model_config = {"extra": "allow"} + + +class LoginConfig(BaseModel): + """Authentication configuration.""" + username: Optional[str] = None + password: Optional[str] = None + + model_config = {"extra": "allow"} + + +class GeneralConfig(BaseModel): + """General settings.""" + version: str = "" + server_update_url: str = Field( + default="https://dashboard.ows.eu/api/owilix/default-cfg-0-17", + alias="server-update-url" + ) + license: Optional[str] = None + + model_config = {"extra": "allow", "populate_by_name": True} + + +class RepositoriesConfig(BaseModel): + """Repositories configuration section.""" + config: Dict[str, RepositoryConfig] = Field(default_factory=dict) + selected_remote: List[str] = Field(default_factory=list) + selected_local: List[str] = Field(default_factory=list) + + model_config = {"extra": "allow"} + + +class OWILIXSettings(BaseModel): + """ + Complete OWILIX configuration model. + + This model represents the full owilix.cfg YAML structure. + + Example YAML: + general: + version: "0.17.0" + server-update-url: "https://..." + repositories: + config: + it4i: + repository: lexis+http + options: + zone: IT4ILexisV2 + selected_remote: [it4i] + selected_local: [local] + profiles: + default: + showfields: "..." + theme: + info: "dim cyan" + """ + general: GeneralConfig = Field(default_factory=GeneralConfig) + repositories: RepositoriesConfig = Field(default_factory=RepositoriesConfig) + profiles: Dict[str, ProfileConfig] = Field(default_factory=dict) + theme: ThemeConfig = Field(default_factory=ThemeConfig) + user: UserConfig = Field(default_factory=UserConfig) + dataset: Dict[str, Any] = Field(default_factory=dict) + login: LoginConfig = Field(default_factory=LoginConfig) + + # Server-provided config (merged at runtime) + server_config: Dict[str, Any] = Field(default_factory=dict, alias="server-config") + + model_config = {"extra": "allow", "populate_by_name": True} + + def get_repository(self, name: str) -> Optional[RepositoryConfig]: + """Get a specific repository configuration.""" + return self.repositories.config.get(name) + + def get_remote_repositories(self) -> Dict[str, RepositoryConfig]: + """Get all selected remote repositories.""" + return { + k: v for k, v in self.repositories.config.items() + if k in self.repositories.selected_remote + } + + def get_local_repositories(self) -> Dict[str, RepositoryConfig]: + """Get all selected local repositories.""" + return { + k: v for k, v in self.repositories.config.items() + if k in self.repositories.selected_local + } + + def get_profile(self, name: str = "default") -> ProfileConfig: + """Get a profile, merging with default.""" + default = self.profiles.get("default", ProfileConfig()) + if name == "default" or name not in self.profiles: + return default + # Merge specific profile over default + profile_data = default.model_dump() + profile_data.update(self.profiles[name].model_dump(exclude_unset=True)) + return ProfileConfig(**profile_data) + + +class OWILIXEnvSettings(BaseSettings): + """ + Environment-based settings for OWILIX. + + These override or supplement file-based configuration. + + Environment Variables: + OWILIX_LEXIS_PROJECT_NAME: Project name + OWILIX_LEXIS_PROJECT_ID: Project ID + OWS_OWI_PATH: Base path for OWILIX data + OWILIX_IRODS_CONNECTION_TIMEOUT: iRODS timeout in seconds + PY4LEXIS_TOKEN: Refresh token for authentication + """ + project_name: str = Field(default="openwebsearch", alias="OWILIX_LEXIS_PROJECT_NAME") + project_id: Optional[str] = Field(default=None, alias="OWILIX_LEXIS_PROJECT_ID") + owi_path: str = Field(default="~/.owi", alias="OWS_OWI_PATH") + irods_connection_timeout: int = Field(default=120, alias="OWILIX_IRODS_CONNECTION_TIMEOUT") + py4lexis_token: Optional[str] = Field(default=None, alias="PY4LEXIS_TOKEN") + + model_config = { + "env_prefix": "", # Use explicit aliases + "extra": "ignore", + "populate_by_name": True, + } + + @field_validator("owi_path") + @classmethod + def expand_path(cls, v: str) -> str: + return os.path.expanduser(v) diff --git a/owilix/core/env.py b/owilix/core/manager/env.py similarity index 73% rename from owilix/core/env.py rename to owilix/core/manager/env.py index 4428d9e..c5e6807 100644 --- a/owilix/core/env.py +++ b/owilix/core/manager/env.py @@ -1,12 +1,19 @@ +""" +OWILIX Environment Configuration + +Provides singleton access to environment variables and configuration settings. +""" + import hashlib import os def _targetProjectHash(project: str) -> str: - + """Generate a project hash from the project name.""" tmpHash: str = hashlib.md5(project.encode("utf8")).hexdigest() return "proj" + tmpHash + class OWILIXEnv: """ Environment configuration class for OWILIX. @@ -15,7 +22,7 @@ class OWILIXEnv: and configuration settings. It handles project name, project ID, OWI path, and connection timeout settings. - The class allows both singleton access through the `instance` property and + The class allows both singleton access through the `values` property and direct instantiation. Properties are read-only to prevent accidental modification. Environment Variables: @@ -26,14 +33,14 @@ class OWILIXEnv: Example: # Using singleton pattern - env = OWILIXEnv.instance + env = OWILIXEnv.values # Direct instantiation env = OWILIXEnv() # Accessing properties print(env.project_name) - print(env.connection_timeout) + print(env.irods_connection_timeout) """ _singleton = None @@ -45,7 +52,7 @@ class OWILIXEnv: self._owi_path = os.path.expanduser(os.getenv("OWS_OWI_PATH", "~/.owi")) # Handle connection timeout - must be integer or None - timeout_str = os.getenv("OWILIX_IRODS_CONNECTION_TIMEOUT", 120) + timeout_str = os.getenv("OWILIX_IRODS_CONNECTION_TIMEOUT", "120") if timeout_str is None or str(timeout_str).strip() == "": self._connection_timeout = None else: @@ -69,47 +76,20 @@ class OWILIXEnv: @property def project_name(self) -> str: - """ - Get the project name. - - Returns: - str: Project name from environment or default - """ + """Get the project name.""" return self._project_name @property def project_id(self) -> str: - """ - Get the project ID. - - Returns: - str: Project ID from environment or auto-generated - """ + """Get the project ID.""" return self._project_id @property def owi_path(self) -> str: - """ - Get the OWI path. - - Returns: - str: Expanded OWI path from environment or default - """ + """Get the OWI path (expanded).""" return self._owi_path - # Keep the old property name for backward compatibility @property def irods_connection_timeout(self) -> int | None: - """ - Get the connection timeout (deprecated property name). - - Returns: - int | None: Connection timeout in seconds or None if not set - - Note: - This property is deprecated. Use connection_timeout instead. - """ + """Get the iRODS connection timeout in seconds.""" return self._connection_timeout - - - diff --git a/owilix/core/logging.py b/owilix/core/manager/logging.py similarity index 99% rename from owilix/core/logging.py rename to owilix/core/manager/logging.py index 7fbeb20..153dc1c 100644 --- a/owilix/core/logging.py +++ b/owilix/core/manager/logging.py @@ -1,3 +1,4 @@ +import hashlib import json import duckdb import logging diff --git a/owilix/core/manager/manager.py b/owilix/core/manager/manager.py new file mode 100644 index 0000000..9fd72b9 --- /dev/null +++ b/owilix/core/manager/manager.py @@ -0,0 +1,526 @@ +""" +OWILiX Manager Module (Refactored) + +This module provides the central orchestration for OWILIX: +- OWIlixConfig: YAML-based configuration with remote updates +- OWIlixManager: Project/session/repository management + +This is a refactored version that integrates with: +- OWILIXSession: Unified session wrapper +- Pydantic models: Type-safe configuration (optional) +""" + +import glob +import json +import logging +import os +import shutil +import stat +import sys +import threading +from datetime import datetime +from typing import Any, Dict, Optional + +import getpass +import dateutil +import pandas as pd +import requests +import yaml + +from owilix._version import __version__ +from owilix.core.manager.env import OWILIXEnv +from owilix.core.utils import dict_multi_key_add, dict_multi_key_get +from owilix.core.manager.session import OWILIXSession + +# Lazy imports to avoid circular dependencies +def _get_repository_classes(): + from owilix.core.repository import ( + LEXISIrodsRepository, + LocalRepository, + AggregatedRepository, + LEXISIrodsHTTPRepository, + S3FileBasedRepository + ) + return { + "lexis": LEXISIrodsRepository, + "lexis+http": LEXISIrodsHTTPRepository, + "s3a": S3FileBasedRepository, + "file": LocalRepository, + }, AggregatedRepository + +logger = logging.getLogger("owilix") + + +def rotate_log_file(filename: str, max_size: int = 500 * 1024): + """Rotate log file if it exceeds max_size.""" + if os.path.isfile(filename): + file_size = os.path.getsize(filename) + if file_size > max_size: + timestamp = datetime.now().strftime('%Y%m%d%H%M%S') + rotated_filename = f"{filename}.{timestamp}" + shutil.copy2(filename, rotated_filename) + open(filename, 'w').close() + + +class OWIlixConfig: + """ + Configuration handler for OWIlix. + + Manages loading, saving, and manipulating YAML-based configurations. + Supports remote server configuration updates and validation. + + Attributes: + config_file: Path to the configuration file. + schema: Optional validation schema. + config: The loaded configuration dictionary. + """ + + def __init__(self, config_file: str, + schema: Dict[str, Any] = None, + no_remote: bool = False): + self.config_file = config_file + self.schema = schema + self.config = self.load_config(no_remote) + + def delete_config(self): + if os.path.exists(self.config_file): + os.remove(self.config_file) + self.config_file = None + self.config = self.load_config() + + def _old_server_urls(self): + return {"https://dashboard.ows.eu/api/owilix/default-cfg"} + + def load_config(self, no_remote: bool = False) -> Dict[str, Any]: + """Load configuration from YAML file with optional remote updates.""" + try: + with open(self.config_file, 'r') as file: + config = yaml.safe_load(file) + _patched_config = self.patch_version(config, os.path.dirname(self.config_file)) + if not no_remote: + if "general" in _patched_config and "server-update-url" in _patched_config["general"]: + _url = _patched_config["general"].get("server-update-url", None) + if _url in self._old_server_urls(): + _patched_config = self.apply_server_config(_patched_config) + else: + _patched_config = self.apply_server_config(_patched_config, _url) + else: + _patched_config = self.apply_server_config(_patched_config) + self.validate_config(_patched_config) + return _patched_config + except FileNotFoundError: + _default = self.get_default(os.path.dirname(self.config_file)) + if not no_remote: + _default = self.apply_server_config(_default) + return _default + + def save_config(self) -> None: + """Save configuration to YAML file.""" + _dir_name = os.path.dirname(self.config_file) + if not os.path.exists(_dir_name): + os.makedirs(_dir_name) + with open(self.config_file, 'w') as file: + yaml.dump(self.config, file) + + def validate_config(self, config: Dict[str, Any]) -> None: + """Validate configuration (placeholder for schema validation).""" + pass + + def patch_version(self, config: Dict[str, Any], target) -> Dict[str, Any]: + """Apply backward compatibility patches.""" + if "version" not in config.get("general", {}): + return self.get_default(target) + return config + + def add_config(self, key: str, value: Any) -> None: + """Add configuration using dot-notation key.""" + self.config = dict_multi_key_add(self.config, key, value) + self.validate_config(self.config) + + def get_config(self, key: str) -> Any: + """Get configuration value by dot-notation key.""" + return dict_multi_key_get(self.config, key) + + def get_repositories(self, remote: list = None, local: list = None): + """Get filtered repository configurations.""" + repo_cfg = self.config.get("repositories", {}) + if remote is None: + remote = repo_cfg.get("selected_remote", []) + elif all([r.startswith("+") for r in remote]): + remote = repo_cfg.get("selected_remote", []) + [k[1:] for k in remote] + else: + remote = [k[1:] for k in remote if k.startswith("+")] + [k for k in remote if not k.startswith("+")] + if local is None: + local = repo_cfg.get("selected_local", []) + + # Patch old config format + for v in repo_cfg.get("config", {}).values(): + _options = v.get("options", {}) + if "name" in _options and "project_name" not in _options: + _options["project_name"] = _options.pop("name") + + return ( + {k: v for k, v in repo_cfg["config"].items() if k in remote}, + {k: v for k, v in repo_cfg["config"].items() if k in local} + ) + + def get_profile(self, name: str = "default"): + """Get display profile configuration.""" + default = dict(self.config.get("profiles", {}).get("default", {})) + if name != "default": + default.update(self.config.get("profiles", {}).get(name, {})) + return default + + def get_theme(self): + """Get theme configuration.""" + return self.config.get("theme", {}) + + def license_agreed(self, license_info): + """Mark license as agreed.""" + if "general" not in self.config: + self.config["general"] = {} + self.config["general"]["license"] = license_info["version"] + self.save_config() + + def apply_server_config(self, local_config: Dict[str, Any], + server_url: str = "https://dashboard.ows.eu/api/owilix/default-cfg-0-17") -> Dict[str, Any]: + """Fetch and merge server configuration.""" + if not server_url: + return local_config + try: + response = requests.get(server_url, timeout=10) + response.raise_for_status() + server_config = response.json() + + if "repositories" in server_config and "config" in server_config["repositories"]: + for repo in server_config["repositories"]["config"].keys(): + local_config.setdefault("repositories", {}).setdefault("config", {})[repo] = \ + server_config["repositories"]["config"][repo] + + if "server-config" in server_config: + local_config["server-config"] = server_config["server-config"] + + return local_config + except requests.RequestException as e: + logger.error(f"Error fetching server config: {e}") + return local_config + + def get_default(self, target) -> Dict[str, Any]: + """Get default configuration.""" + return { + "general": { + "server-update-url": "https://dashboard.ows.eu/api/owilix/default-cfg-0-17", + "version": str(__version__) + }, + "repositories": { + "config": { + "it4i": { + "repository": "lexis+http", + "options": { + "project_name": OWILIXEnv.values.project_name, + "lexis_id": OWILIXEnv.values.project_id, + "zone": "IT4ILexisV2" + } + }, + "lrz": { + "repository": "lexis+http", + "options": { + "project_name": OWILIXEnv.values.project_name, + "lexis_id": OWILIXEnv.values.project_id, + "zone": "IT4ILexisV2", + "zone_path": "OWSLRZZONE" + } + }, + "local": { + "repository": "file", + "options": { + "path": target + os.path.sep + "{access}", + "collections_in_path": True + } + } + }, + "selected_remote": ["it4i", "lrz"], + "selected_local": ["local"] + }, + "profiles": { + "default": { + "showfields": "internalID,title,dataCenter,startDate,endDate,resourceType,access,collectionName", + "nodisplay": ".*", + "sort": "+startDate", + }, + "mini": { + "showfields": "internalID,title,dataCenter", + "nodisplay": ".*", + "sort": "+startDate", + } + }, + "theme": { + "info": "dim cyan", + "warning": "magenta", + "danger": "bold red", + "error": "bold red" + }, + "user": { + "name": getpass.getuser(), + "email": "", + "organisation": "" + }, + "dataset": {"metadata": {}} + } + + +class OWIlixManager: + """ + Central orchestrator for OWILIX operations. + + Manages: + - Configuration loading and persistence + - Session authentication (via OWILIXSession) + - Repository instantiation and aggregation + - Logging and statistics + + Attributes: + name: Project name (from environment or default). + config: OWIlixConfig instance. + session: OWILIXSession for authentication and iRODS access. + local: Local repository instance. + remote_data: Aggregated remote repositories. + """ + + def __init__(self, + owi_path: str = None, + logpath: str = None, + repos: Dict[str, Dict] = None, + name: str = "openwebsearch", + user: str = None, + config: OWIlixConfig = None, + no_refresh_token: bool = False): + """ + Initialize the manager. + + Args: + owi_path: Base path for OWILIX data (default: ~/.owi). + logpath: Path for log files (default: owi_path/.logs). + repos: Manual repository configurations (overrides config). + name: Project name. + user: User name (overrides environment). + config: Pre-loaded configuration object. + no_refresh_token: Force fresh authentication. + """ + # Use environment defaults + owi_path = owi_path or OWILIXEnv.values.owi_path + + self.repos = repos + self.no_refresh_token = no_refresh_token + self.name = OWILIXEnv.values.project_name + self.lexis_id = OWILIXEnv.values.project_id + + # Setup logging + self.logger = logging.getLogger("owilix") + if not self.logger.handlers: + handler = logging.StreamHandler(sys.stderr) + formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') + handler.setFormatter(formatter) + self.logger.addHandler(handler) + + # Paths + self.owi_path = owi_path + self.logpath = logpath or os.path.join(owi_path, ".logs") + self._configpath = owi_path + + if not os.path.exists(self._configpath): + os.makedirs(self._configpath) + + self.config = config or OWIlixConfig(os.path.join(self._configpath, "owilix.cfg")) + + self._token_path = os.path.join(self._configpath, ".tokens") + if not os.path.exists(self._token_path): + os.makedirs(self._token_path) + + # Load repositories from config if not provided + if not repos: + remote, local = self.config.get_repositories() + repos = {**remote, **local} + + if not os.path.exists(self.logpath): + os.makedirs(self.logpath) + + self.logfile = os.path.join(self.logpath, "events.json") + self._repo_stats_file = os.path.join(self.logpath, "repository_stats.json") + rotate_log_file(self.logfile) + + # Initialize repositories + repo_classes, AggregatedRepository = _get_repository_classes() + _repos = {} + self._local = None + self._user = user + + for repo_name, repo_config in repos.items(): + repo_type = repo_config["repository"] + if repo_type not in repo_classes: + raise ValueError(f"Unknown repository type: {repo_type}") + + RepoClass = repo_classes[repo_type] + repo = RepoClass(self, repo_name=repo_name, **repo_config["options"]) + + if repo_name == "local": + self._local = repo + else: + _repos[repo_name] = repo + + self._remote_data = AggregatedRepository(_repos) + self._remote_data.load_stats_from_file(self._repo_stats_file) + + if self._local is None: + raise ValueError("No local repository defined in configuration") + + # Session management + self._session = None + self._owilix_session = None + self._session_lock = threading.Lock() + + @property + def _refresh_token_fn(self) -> str: + return os.path.join(self._token_path, "refresh_token") + + def clean(self, remove_cfg: bool = False): + """Clean tokens and optionally configuration.""" + if os.path.exists(self._refresh_token_fn): + os.remove(self._refresh_token_fn) + if remove_cfg: + self.config.delete_config() + + @property + def local(self): + """Local repository instance.""" + return self._local + + @property + def user(self) -> Optional[str]: + """Current user name.""" + return self._user + + @property + def remote_data(self): + """Aggregated remote repositories.""" + return self._remote_data + + @property + def session(self) -> OWILIXSession: + """ + Get the unified session wrapper (thread-safe, lazy-loaded). + + Returns: + OWILIXSession: Session with access to LexisSession and iRODS. + """ + if self._owilix_session is None: + with self._session_lock: + if self._owilix_session is None: + lexis_session = self._get_lexis_session() + self._owilix_session = OWILIXSession(lexis_session) + return self._owilix_session + + @property + def lexis_session(self): + """ + Direct access to LexisSession (for backward compatibility). + + Returns: + LexisSession: The underlying LEXIS session. + """ + return self.session.lexis + + def _get_lexis_session(self): + """Create or retrieve LexisSession with authentication.""" + from py4lexis.session import LexisSession + from py4lexis.core.exceptions import Py4LexisAuthException + + # Try credentials from config + _login = self.config.get_config("login") or {} + if _login.get("username"): + self.logger.info("Using credentials from configuration") + return LexisSession( + suppress_print=True, in_cli=True, + login_method="credentials", + log_file=self.get_lexis_log_filename(), + username=_login.get("username"), + password=_login.get("password") + ) + + # Try refresh token + _refresh_token = os.environ.get("PY4LEXIS_TOKEN") or self._refresh_token_io() + _refresh_token = None if self.no_refresh_token else _refresh_token + + try: + if _refresh_token: + session = LexisSession( + suppress_print=True, in_cli=True, + offline_access=True, + login_method="offline", + log_file=self.get_lexis_log_filename(), + refresh_token=_refresh_token + ) + if session.get_access_token() is None: + raise Py4LexisAuthException("Token expired") + else: + raise Py4LexisAuthException("No refresh token") + except Py4LexisAuthException: + # Fallback to URL login + session = LexisSession( + suppress_print=False, in_cli=True, + offline_access=True, + login_method="url", + log_file=self.get_lexis_log_filename() + ) + + # Persist new token + _new_token = session.get_offline_token() or session.get_refresh_token() + if _new_token and _new_token != _refresh_token: + self._refresh_token_io(_new_token) + + return session + + def _refresh_token_io(self, token: str = None) -> Optional[str]: + """Read or write refresh token with secure permissions.""" + if token is None: + try: + with open(self._refresh_token_fn, 'r') as file: + return file.read() + except FileNotFoundError: + return None + else: + with open(self._refresh_token_fn, 'w') as file: + file.write(token) + os.chmod(self._refresh_token_fn, stat.S_IRUSR | stat.S_IWUSR) + return token + + def update_log(self, data_dict: dict): + """Append event to log file.""" + data_dict["logdate"] = str(datetime.now().strftime('%Y-%m-%d %H:%M:%S')) + with open(self.logfile, "a") as file: + file.write(json.dumps(data_dict) + '\n') + + def load_log(self, all: bool = True) -> list: + """Load log entries from JSON files.""" + data = [] + log_files = glob.glob(os.path.join(self.logpath, "events*.json")) + for log_file in log_files: + with open(log_file, 'r') as file: + for line in file: + if line.strip(): + data.append(json.loads(line.strip())) + return data + + def get_lexis_log_filename(self) -> str: + return os.path.join(self.logpath, "lexis.log") + + def get_error_log_filename(self) -> str: + return os.path.join(self.logpath, "errors.log") + + def save_repository_stats(self): + """Persist repository performance statistics.""" + self._remote_data.store_stats_to_file(self._repo_stats_file) + + +# Backward compatibility alias +OWILIXManager = OWIlixManager diff --git a/owilix/core/manager/session.py b/owilix/core/manager/session.py new file mode 100644 index 0000000..3a2e457 --- /dev/null +++ b/owilix/core/manager/session.py @@ -0,0 +1,140 @@ +""" +OWILIX Session Wrapper + +Provides unified access to LexisSession and iRODS capabilities. +The session wrapper ensures proper token refresh and lazy initialization. +""" + +import logging +from typing import TYPE_CHECKING, Optional + +if TYPE_CHECKING: + from py4lexis.session import LexisSession + from irods_http_client import IrodsHttpClient + +logger = logging.getLogger("owilix.session") + + +class OWILIXSession: + """ + Unified session wrapper for LEXIS and iRODS access. + + This class provides lazy-loaded access to: + - LexisSession: Authentication and LEXIS API access + - iRODS: iRODS operations wrapper (py4lexis.core.lexis_irods.iRODS) + - IrodsHttpClient: Direct HTTP client for iRODS operations + + The iRODS wrapper is only loaded when first accessed, allowing + the session to be used without python-irodsclient installed. + + Example: + session = OWILIXSession(lexis_session) + + # Access LEXIS API + token = session.lexis.get_access_token() + + # Access iRODS (lazy-loaded) + irods = session.irods + + # Get HTTP client with fresh token + client = session.get_irods_client() + """ + + def __init__(self, lexis_session: "LexisSession"): + """ + Initialize the session wrapper. + + Args: + lexis_session: An authenticated LexisSession instance. + """ + self._session = lexis_session + self._irods = None + self._irods_available = None # Cached availability check + + @property + def lexis(self) -> "LexisSession": + """ + Get the underlying LexisSession. + + Returns: + LexisSession: The LEXIS session for API access. + """ + return self._session + + @property + def irods_available(self) -> bool: + """ + Check if py4lexis iRODS support is available. + + Returns: + bool: True if iRODS wrapper can be used. + """ + if self._irods_available is None: + try: + from py4lexis.core.lexis_irods import iRODS + self._irods_available = True + except ImportError: + self._irods_available = False + logger.debug("py4lexis iRODS support not available") + return self._irods_available + + @property + def irods(self): + """ + Get the iRODS wrapper (lazy-loaded). + + Returns: + iRODS: The py4lexis iRODS wrapper. + + Raises: + ImportError: If py4lexis iRODS support is not installed. + """ + if self._irods is None: + from py4lexis.core.lexis_irods import iRODS + self._irods = iRODS(self._session, suppress_print=True) + logger.debug("Initialized iRODS wrapper") + return self._irods + + def get_irods_client(self) -> "IrodsHttpClient": + """ + Get the IrodsHttpClient with a valid (refreshed) token. + + This method ensures the access token is valid before returning + the HTTP client, handling refresh automatically. + + Returns: + IrodsHttpClient: HTTP client for iRODS operations. + """ + # Call private method to check/refresh token + self.irods._iRODS__check_access_token() + return self.irods._irds + + def get_access_token(self) -> Optional[str]: + """ + Get the current access token. + + Returns: + str: The access token, or None if not authenticated. + """ + return self._session.get_access_token() + + def get_refresh_token(self) -> Optional[str]: + """ + Get the refresh token for persistent authentication. + + Returns: + str: The refresh token, or None if not available. + """ + return self._session.get_refresh_token() + + @property + def url_base(self) -> Optional[str]: + """ + Get the iRODS HTTP API base URL. + + Returns: + str: The base URL for iRODS HTTP API, or None if not available. + """ + if self._irods is not None: + return self._irods._irds.url_base if hasattr(self._irods, "_irds") else None + return None diff --git a/owilix/core/ui.py b/owilix/core/manager/ui.py similarity index 99% rename from owilix/core/ui.py rename to owilix/core/manager/ui.py index f5c6885..2607b57 100644 --- a/owilix/core/ui.py +++ b/owilix/core/manager/ui.py @@ -17,7 +17,7 @@ from rich.table import Table from rich.text import Text from owilix.core.utils import OwilixJSONEncoder -from owilix.core.logging import * +from owilix.core.manager.logging import * import json import logging import sys diff --git a/pyproject.toml b/pyproject.toml index 5832d6f..778a401 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,6 +27,7 @@ dependencies = [ "duckdb>1.0.0", "ipython>8.24.0", "pydantic>=2.5.3,<3.0.0", + "pydantic-settings>=2.0.0", "rich>=13.7.0", "numpy>2.0.0", "typer>=0.13.1", @@ -45,6 +46,9 @@ owilix = "owilix.cli:main" [project.optional-dependencies] plugins = [] +irods-tcp = [ + "python-irodsclient>=2.0.0", +] # ---------------------- uv settings ---------------------- [tool.uv] diff --git a/tests/check_http2.py b/tests/check_http2.py new file mode 100644 index 0000000..996a57f --- /dev/null +++ b/tests/check_http2.py @@ -0,0 +1,43 @@ + +import asyncio +import httpx +import os +from py4lexis.session import LexisSession +from py4lexis.core.lexis_irods import iRODS + +class OWIIrods(iRODS): + def get_client(self): + self._iRODS__check_access_token() + return self._irds + +async def check_http2(): + # reused session logic + token_path = os.path.expanduser("~/tmp/refresh_token.txt") + try: + with open(token_path, "r") as f: + refresh_token = f.read().strip() + session = LexisSession(login_method="token", refresh_token=refresh_token) + except FileNotFoundError: + session = LexisSession() + + irods_wrapper = OWIIrods(session=session) + client = irods_wrapper.get_client() + # iClient has .token property according to previous analysis of collection_operations.py + # collection_operations.py: self.sess.token + # irodsHttpClient (client): has .token property + + token = client.token + url_base = client.url_base + + print(f"Checking HTTP/2 support for: {url_base}") + + async with httpx.AsyncClient(http2=True, verify=False) as http_client: + try: + response = await http_client.get(f"{url_base}/info", headers={"Authorization": f"Bearer {token}"}) + print(f"Protocol: {response.http_version}") + print(f"Status: {response.status_code}") + except Exception as e: + print(f"Error: {e}") + +if __name__ == "__main__": + asyncio.run(check_http2()) diff --git a/tests/lexis_session_test.py b/tests/lexis_session_test.py deleted file mode 100644 index 9e36d67..0000000 --- a/tests/lexis_session_test.py +++ /dev/null @@ -1,22 +0,0 @@ -import datetime -import py4lexis -from py4lexis.session import LexisSession -from time import sleep - -session = LexisSession() -print("Lexis Version", py4lexis.__version__) - -old_token = session._tokens.refresh_token -print(session.refresh_token) -retrieved_at =session._tokens.token_retrieved_at -original_datetime = datetime.datetime(2024, 1, 1, 0, 0) + datetime.timedelta(seconds=session._tokens.token_retrieved_at) -print("Retrieved at ", original_datetime, " refresh expires ", session._tokens.refresh_expires_in, " expires in", session._tokens.expires_in ) -sleep(90) -print("WAITING 90") -session.refresh_token(session.get_refresh_token()) -original_datetime_2 = datetime.datetime(2024, 1, 1, 0, 0) + datetime.timedelta(seconds=session._tokens.token_retrieved_at) -print("Retrieved at ", original_datetime_2, " refresh expires ", session._tokens.refresh_expires_in, " expires in", session._tokens.expires_in ) -if session._tokens.refresh_token != old_token: - print("!!!! Session token different") - if retrieved_at==session._tokens.token_retrieved_at: - print("but retrieved time is the same!!!!") \ No newline at end of file diff --git a/tests/owilix/core/fsspec/benchmark_results.jsonl b/tests/owilix/core/fsspec/benchmark_results.jsonl new file mode 100644 index 0000000..3741b6c --- /dev/null +++ b/tests/owilix/core/fsspec/benchmark_results.jsonl @@ -0,0 +1,6 @@ +{"timestamp": "2026-01-01T16:07:46.999990", "method": "async_httpx", "file": "metadata_0.parquet", "size_bytes": 157859, "duration_sec": 0.22817206382751465, "speed_mb_s": 0.6597918754303711, "chunk_size": 1048576} +{"timestamp": "2026-01-01T16:07:47.233134", "method": "async_httpx", "file": "metadata_0.parquet", "size_bytes": 166640, "duration_sec": 0.2328348159790039, "speed_mb_s": 0.6825452087898585, "chunk_size": 1048576} +{"timestamp": "2026-01-01T16:07:47.451806", "method": "async_httpx", "file": "metadata_0.parquet", "size_bytes": 113661, "duration_sec": 0.2184619903564453, "speed_mb_s": 0.4961759082217973, "chunk_size": 1048576} +{"timestamp": "2026-01-01T16:07:47.665334", "method": "async_httpx", "file": "metadata_0.parquet", "size_bytes": 54638, "duration_sec": 0.21330618858337402, "speed_mb_s": 0.24428197627954856, "chunk_size": 1048576} +{"timestamp": "2026-01-01T16:07:47.873873", "method": "async_httpx", "file": "metadata_0.parquet", "size_bytes": 15990, "duration_sec": 0.2083420753479004, "speed_mb_s": 0.07319333981804657, "chunk_size": 1048576} +{"timestamp": "2026-01-01T16:07:54.920395", "method": "async_httpx_concurrent", "file": "multiple_10", "size_bytes": 39527058, "duration_sec": 7.046247959136963, "speed_mb_s": 5.34978902762276, "chunk_size": 1048576} diff --git a/tests/owilix/core/fsspec/benchmark_results_20260101_130824.json b/tests/owilix/core/fsspec/benchmark_results_20260101_130824.json deleted file mode 100644 index ec79bb8..0000000 --- a/tests/owilix/core/fsspec/benchmark_results_20260101_130824.json +++ /dev/null @@ -1,60 +0,0 @@ -[ - { - "name": "File listing (python-irodsclient)", - "duration_seconds": 86.402788041858, - "items_count": 1524, - "bytes_processed": 98027926389, - "throughput_mb_s": 1081.9872576618934, - "parameters": {} - }, - { - "name": "Sequential read (chunk=256KB)", - "duration_seconds": 1.3575281659141183, - "items_count": 0, - "bytes_processed": 2097152, - "throughput_mb_s": 1.4732659330521225, - "parameters": { - "chunk_size": 262144 - } - }, - { - "name": "Sequential read (chunk=1024KB)", - "duration_seconds": 0.6293834999669343, - "items_count": 0, - "bytes_processed": 2097152, - "throughput_mb_s": 3.177712793718096, - "parameters": { - "chunk_size": 1048576 - } - }, - { - "name": "Sequential read (chunk=4096KB)", - "duration_seconds": 1.5911380001343787, - "items_count": 0, - "bytes_processed": 4194304, - "throughput_mb_s": 2.5139239963235007, - "parameters": { - "chunk_size": 4194304 - } - }, - { - "name": "Seek+read (size=64KB)", - "duration_seconds": 0.3701308339368552, - "items_count": 0, - "bytes_processed": 131072, - "throughput_mb_s": 0.33771841883706766, - "parameters": { - "read_size": 65536 - } - }, - { - "name": "Seek+read (size=256KB)", - "duration_seconds": 0.49396199989132583, - "items_count": 0, - "bytes_processed": 524288, - "throughput_mb_s": 1.0122236125653443, - "parameters": { - "read_size": 262144 - } - } -] \ No newline at end of file diff --git a/tests/owilix/core/manager/__init__.py b/tests/owilix/core/manager/__init__.py new file mode 100644 index 0000000..4e41397 --- /dev/null +++ b/tests/owilix/core/manager/__init__.py @@ -0,0 +1 @@ +# Tests for owilix.core.manager package diff --git a/tests/owilix/core/manager/test_logging.py b/tests/owilix/core/manager/test_logging.py new file mode 100644 index 0000000..59264a4 --- /dev/null +++ b/tests/owilix/core/manager/test_logging.py @@ -0,0 +1,97 @@ +""" +Tests for owilix.core.manager.logging module + +Tests for DBLog transaction logging system. +""" + +import tempfile +import pytest + + +class TestDBLog: + """Tests for DBLog transaction logging system.""" + + @pytest.fixture + def temp_log_dir(self): + """Create temporary directory for test logs.""" + with tempfile.TemporaryDirectory() as tmpdir: + yield tmpdir + + def test_dblog_initialization(self, temp_log_dir): + """Test DBLog can be initialized.""" + from owilix.core.manager.logging import DBLog + + dblog = DBLog("test_job", temp_log_dir) + assert dblog is not None + assert dblog.job_name == "test_job" + + def test_dblog_register_dataset(self, temp_log_dir): + """Test registering a dataset.""" + from owilix.core.manager.logging import DBLog + + dblog = DBLog("test_job", temp_log_dir) + dataset_id = dblog.register_dataset("/path/to/dataset") + assert dataset_id is not None + assert isinstance(dataset_id, str) + assert len(dataset_id) == 16 # SHA256 truncated to 16 chars + + def test_dblog_register_files(self, temp_log_dir): + """Test registering files for processing.""" + from owilix.core.manager.logging import DBLog + + dblog = DBLog("test_job", temp_log_dir) + # API expects (file_path, dataset_path) tuples + files = [ + ("/path/to/file1.parquet", "/path/to/dataset"), + ("/path/to/file2.parquet", "/path/to/dataset"), + ] + dblog.register_files(files) + + stats = dblog.get_statistics() + assert stats["file_status_counts"].get("pending", 0) == 2 + + def test_dblog_mark_success(self, temp_log_dir): + """Test marking files as successfully processed.""" + from owilix.core.manager.logging import DBLog + + dblog = DBLog("test_job", temp_log_dir) + files = [("/path/to/file1.parquet", "/path/to/dataset")] + dblog.register_files(files) + dblog.mark_files_success(["/path/to/file1.parquet"]) + + processed = dblog.get_processed_files() + assert "/path/to/file1.parquet" in processed + + def test_dblog_mark_error(self, temp_log_dir): + """Test marking files as failed.""" + from owilix.core.manager.logging import DBLog + + dblog = DBLog("test_job", temp_log_dir) + files = [("/path/to/file1.parquet", "/path/to/dataset")] + dblog.register_files(files) + + error = ValueError("Test error") + dblog.mark_files_error(["/path/to/file1.parquet"], error) + + stats = dblog.get_statistics() + assert stats["file_status_counts"].get("error", 0) == 1 + + def test_dblog_get_statistics(self, temp_log_dir): + """Test getting processing statistics.""" + from owilix.core.manager.logging import DBLog + + dblog = DBLog("test_job", temp_log_dir) + stats = dblog.get_statistics() + + assert "file_status_counts" in stats + assert "dataset_statistics" in stats + assert "error_type_counts" in stats + + +class TestDBLogImport: + """Test DBLog can be imported from manager package.""" + + def test_import_from_manager_logging(self): + """Test import from manager.logging.""" + from owilix.core.manager.logging import DBLog + assert DBLog is not None diff --git a/tests/owilix/core/manager/test_manager.py b/tests/owilix/core/manager/test_manager.py new file mode 100644 index 0000000..8937f5b --- /dev/null +++ b/tests/owilix/core/manager/test_manager.py @@ -0,0 +1,124 @@ +""" +Tests for owilix.core.manager package + +This test module covers: +- OWILIXEnv environment configuration +- OWILIXSettings Pydantic models +- OWILIXSession wrapper +- OWILIXManager initialization +""" + +import os +import pytest +from unittest.mock import patch, MagicMock + + +class TestOWILIXEnv: + """Tests for OWILIXEnv environment configuration.""" + + def test_env_singleton(self): + """Test that OWILIXEnv provides a singleton.""" + from owilix.core.manager import OWILIXEnv + + env1 = OWILIXEnv.values + env2 = OWILIXEnv.values + assert env1 is env2 + + def test_env_default_values(self): + """Test default environment values.""" + from owilix.core.manager import OWILIXEnv + + env = OWILIXEnv.values + assert env.project_name == "openwebsearch" + assert env.owi_path.endswith(".owi") or ".owi" in env.owi_path + assert isinstance(env.irods_connection_timeout, int) + + @patch.dict(os.environ, {"OWILIX_LEXIS_PROJECT_NAME": "test_project"}) + def test_env_respects_environment_variable(self): + """Test that OWILIXEnv reads from environment.""" + from owilix.core.manager.env import OWILIXEnv + + # Clear singleton to force re-read + OWILIXEnv._singleton = None + env = OWILIXEnv() + assert env.project_name == "test_project" + + +class TestOWILIXSettings: + """Tests for Pydantic configuration models.""" + + def test_settings_default(self): + """Test OWILIXSettings with default values.""" + from owilix.core.manager import OWILIXSettings + + settings = OWILIXSettings() + assert settings.theme is not None + assert settings.profiles is not None # Empty dict is valid + + def test_repository_config_validation(self): + """Test RepositoryConfig validation.""" + from owilix.core.manager import RepositoryConfig + + # Valid config + config = RepositoryConfig(repository="lexis+http", options={"zone": "test"}) + assert config.repository == "lexis+http" + assert config.options["zone"] == "test" + + def test_repository_config_invalid_type(self): + """Test RepositoryConfig rejects invalid repository types.""" + from owilix.core.manager import RepositoryConfig + from pydantic import ValidationError + + with pytest.raises(ValidationError): + RepositoryConfig(repository="invalid_type", options={}) + + +class TestOWILIXSession: + """Tests for OWILIXSession wrapper.""" + + def test_session_wraps_lexis_session(self): + """Test OWILIXSession wraps a LexisSession.""" + from owilix.core.manager import OWILIXSession + + mock_lexis = MagicMock() + mock_lexis.get_access_token.return_value = "test_token" + + session = OWILIXSession(mock_lexis) + assert session.lexis is mock_lexis + assert session.get_access_token() == "test_token" + + def test_session_irods_lazy_load(self): + """Test iRODS is not loaded until accessed.""" + from owilix.core.manager import OWILIXSession + + mock_lexis = MagicMock() + session = OWILIXSession(mock_lexis) + + # iRODS should not be loaded yet + assert session._irods is None + + +class TestBackwardCompatibility: + """Tests for backward compatibility with core/__init__.py exports.""" + + def test_env_import_from_manager_package(self): + """Test OWILIXEnv can be imported from manager package.""" + from owilix.core.manager import OWILIXEnv + assert OWILIXEnv is not None + + def test_manager_import_from_core(self): + """Test OWIlixManager can be imported from core.""" + from owilix.core import OWIlixManager + assert OWIlixManager is not None + + def test_ui_import_from_manager_package(self): + """Test UI components can be imported from manager package.""" + from owilix.core.manager import OWILIXConsole, ErrorCollector + assert OWILIXConsole is not None + assert ErrorCollector is not None + + + def test_load_and_check_function(self): + """Test load_and_check utility function.""" + from owilix.core.manager import load_and_check + assert callable(load_and_check) diff --git a/tests/small_tst_py4lexis.py b/tests/small_tst_py4lexis.py index 1c216af..ff5d5e3 100644 --- a/tests/small_tst_py4lexis.py +++ b/tests/small_tst_py4lexis.py @@ -1,22 +1,66 @@ from py4lexis.core.lexis_irods import iRODS from py4lexis.session import LexisSession from py4lexis.ddi.datasets import Datasets +import os -session = LexisSession() + +# read refresh token from file if file exists +_token_path = os.path.expanduser("~/tmp/refresh_token.txt") +try: + with open(_token_path, "r") as f: + _refresh_token = f.read() +except FileNotFoundError: + _refresh_token = None + +if _refresh_token: + session = LexisSession(login_method="token",refresh_token=_refresh_token) +else: + session = LexisSession() + +#store refrehs token in ~/tmp/refresh_token.txt +with open(_token_path, "w") as f: + f.write(session.get_refresh_token()) # --------------------------------------------------------------------------------------------------------------------------------------------------------------------------- # Manage LEXIS datasets WHERE url like '%impressum%' OR url like '%legal%' OR url like '%imprint%' OR url like '%terms%' OR url like '%privacy%' OR url like '%contact%' OR url like '%agreement%' # --------------------------------------------------------------------------------------------------------------------------------------------------------------------------- # Get Datasets manager + +class OWIIrods(iRODS): + + def __init__(self, *args, **kwargs ): + super().__init__(*args, **kwargs) + + def irods(self): # we need this to check for refresh. Unfortunately not + self._iRODS__check_access_token() + return self._irds + ds = Datasets(session, suppress_print=False) -print("getting datasets") -print(ds.get_all_datasets()) -print("downloading dataset with irods") -irods = iRODS(session=session, +slow=False +if slow: + print("getting datasets") + print(ds.get_all_datasets()) + print("downloading dataset with irods") + +irods = OWIIrods(session=session, suppress_print=False) -irods.get_dataset_collection("public", "openwebsearch", "f6ea5756-2e0b-11ef-b336-0242ac1d0004") +coll = irods.get_dataset_collection("f6ea5756-2e0b-11ef-b336-0242ac1d0004") +print("data objects", coll.data_objects) +print("subcollections", coll.subcollections) + +# Collections and DataObjects are more powerful operation, allowing e.g. range requests. But require handling parallel requests. +from irods_http_client.collection_operations import Collections +from irods_http_client.data_object_operations import DataObjects +print("collection stats", Collections(irods.irods(), url_base=irods.irods().url_base).stat(coll.path)) +print("data object stats", DataObjects(irods.irods(), url_base=irods.irods().url_base).stat(coll.data_objects[0].path)) -print("downloading dataset") -ds.download_dataset(dataset_id="f6ea5756-2e0b-11ef-b336-0242ac1d0004") +print("data object content with file path", DataObjects(irods.irods(), url_base=irods.irods().url_base).read(coll.data_objects[0].path, offset=10, count=20)) +# here we use the iRODSDataObject which wraps parts using multithreading. To be prefered when downloading large files +# also has async writer and reader. get_async_reader(), get_async_writer() +print("data object content full (use irodsdataobject, which wraps multithreading)", coll.data_objects[0]) +with coll.data_objects[0].open() as f: + print(f.read()) +#print("downloading dataset") +#ds.download_dataset(dataset_id="f6ea5756-2e0b-11ef-b336-0242ac1d0004") diff --git a/uv.lock b/uv.lock index e2c7bec..c785c00 100644 --- a/uv.lock +++ b/uv.lock @@ -928,6 +928,27 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/31/b4/b9b800c45527aadd64d5b442f9b932b00648617eb5d63d2c7a6587b7cafc/jmespath-1.0.1-py3-none-any.whl", hash = "sha256:02e2e4cc71b5bcab88332eebf907519190dd9e6e82107fa7f83b1003a6252980", size = 20256, upload-time = "2022-06-17T18:00:10.251Z" }, ] +[[package]] +name = "jsonpatch" +version = "1.33" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "jsonpointer" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/42/78/18813351fe5d63acad16aec57f94ec2b70a09e53ca98145589e185423873/jsonpatch-1.33.tar.gz", hash = "sha256:9fcd4009c41e6d12348b4a0ff2563ba56a2923a7dfee731d004e212e1ee5030c", size = 21699, upload-time = "2023-06-26T12:07:29.144Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/73/07/02e16ed01e04a374e644b575638ec7987ae846d25ad97bcc9945a3ee4b0e/jsonpatch-1.33-py2.py3-none-any.whl", hash = "sha256:0ae28c0cd062bbd8b8ecc26d7d164fbbea9652a1a3693f3b956c1eae5145dade", size = 12898, upload-time = "2023-06-16T21:01:28.466Z" }, +] + +[[package]] +name = "jsonpointer" +version = "3.0.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/6a/0a/eebeb1fa92507ea94016a2a790b93c2ae41a7e18778f85471dc54475ed25/jsonpointer-3.0.0.tar.gz", hash = "sha256:2b2d729f2091522d61c3b31f82e11870f60b68f43fbc705cb76bf4b832af59ef", size = 9114, upload-time = "2024-06-10T19:24:42.462Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/71/92/5e77f98553e9e75130c78900d000368476aed74276eb8ae8796f65f00918/jsonpointer-3.0.0-py2.py3-none-any.whl", hash = "sha256:13e088adc14fca8b6aa8177c044e12701e6ad4b28ff10e65f2267a90109c9942", size = 7595, upload-time = "2024-06-10T19:24:40.698Z" }, +] + [[package]] name = "jsonschema" version = "4.25.1" @@ -1187,6 +1208,7 @@ dependencies = [ { name = "py4lexis" }, { name = "pyarrow" }, { name = "pydantic" }, + { name = "pydantic-settings" }, { name = "python-dateutil" }, { name = "python-http-irods-client" }, { name = "pyyaml" }, @@ -1199,6 +1221,11 @@ dependencies = [ { name = "url-normalize" }, ] +[package.optional-dependencies] +irods-tcp = [ + { name = "python-irodsclient" }, +] + [package.dev-dependencies] bloom = [ { name = "datasketch" }, @@ -1254,8 +1281,10 @@ requires-dist = [ { name = "py4lexis", specifier = ">=5.0.0", index = "https://opencode.it4i.eu/api/v4/projects/107/packages/pypi/simple" }, { name = "pyarrow", specifier = "<20.0.0" }, { name = "pydantic", specifier = ">=2.5.3,<3.0.0" }, + { name = "pydantic-settings", specifier = ">=2.0.0" }, { name = "python-dateutil", specifier = ">=2.9.0.post0" }, { name = "python-http-irods-client", git = "https://opencode.it4i.eu/lexis-platform/data/python-http-irods-client.git?rev=1.1.1" }, + { name = "python-irodsclient", marker = "extra == 'irods-tcp'", specifier = ">=2.0.0" }, { name = "pyyaml", specifier = ">=6.0.1,<7.0.0" }, { name = "requests", specifier = ">=2.32.3" }, { name = "rich", specifier = ">=13.7.0" }, @@ -1265,7 +1294,7 @@ requires-dist = [ { name = "typer", specifier = ">=0.13.1" }, { name = "url-normalize", specifier = ">=2.2.1" }, ] -provides-extras = ["plugins"] +provides-extras = ["plugins", "irods-tcp"] [package.metadata.requires-dev] bloom = [ @@ -1585,6 +1614,20 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/32/56/8a7ca5d2cd2cda1d245d34b1c9a942920a718082ae8e54e5f3e5a58b7add/pydantic_core-2.33.2-pp311-pypy311_pp73-win_amd64.whl", hash = "sha256:329467cecfb529c925cf2bbd4d60d2c509bc2fb52a20c1045bf09bb70971a9c1", size = 2066757, upload-time = "2025-04-23T18:33:30.645Z" }, ] +[[package]] +name = "pydantic-settings" +version = "2.12.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pydantic" }, + { name = "python-dotenv" }, + { name = "typing-inspection" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/43/4b/ac7e0aae12027748076d72a8764ff1c9d82ca75a7a52622e67ed3f765c54/pydantic_settings-2.12.0.tar.gz", hash = "sha256:005538ef951e3c2a68e1c08b292b5f2e71490def8589d4221b95dab00dafcfd0", size = 194184, upload-time = "2025-11-10T14:25:47.013Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c1/60/5d4751ba3f4a40a6891f24eec885f51afd78d208498268c734e256fb13c4/pydantic_settings-2.12.0-py3-none-any.whl", hash = "sha256:fddb9fd99a5b18da837b29710391e945b1e30c135477f484084ee513adb93809", size = 51880, upload-time = "2025-11-10T14:25:45.546Z" }, +] + [[package]] name = "pygments" version = "2.18.0" @@ -1718,6 +1761,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ec/57/56b9bcc3c9c6a792fcbaf139543cee77261f3651ca9da0c93f5c1221264b/python_dateutil-2.9.0.post0-py2.py3-none-any.whl", hash = "sha256:a8b2bc7bffae282281c8140a97d3aa9c14da0b136dfe83f850eea9a5f7470427", size = 229892, upload-time = "2024-03-01T18:36:18.57Z" }, ] +[[package]] +name = "python-dotenv" +version = "1.2.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f0/26/19cadc79a718c5edbec86fd4919a6b6d3f681039a2f6d66d14be94e75fb9/python_dotenv-1.2.1.tar.gz", hash = "sha256:42667e897e16ab0d66954af0e60a9caa94f0fd4ecf3aaf6d2d260eec1aa36ad6", size = 44221, upload-time = "2025-10-26T15:12:10.434Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/14/1b/a298b06749107c305e1fe0f814c6c74aea7b2f1e10989cb30f544a1b3253/python_dotenv-1.2.1-py3-none-any.whl", hash = "sha256:b81ee9561e9ca4004139c6cbba3a238c32b03e4894671e181b671e8cb8425d61", size = 21230, upload-time = "2025-10-26T15:12:09.109Z" }, +] + [[package]] name = "python-gitlab" version = "4.13.0" @@ -1744,16 +1796,17 @@ dependencies = [ [[package]] name = "python-irodsclient" -version = "1.1.5" +version = "3.2.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "defusedxml" }, + { name = "jsonpatch" }, + { name = "jsonpointer" }, { name = "prettytable" }, - { name = "six" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/a5/53/f62829a28bb7ba54a43699580ad3b7fbdc2fda4084e49b935416cf8f4bf9/python-irodsclient-1.1.5.tar.gz", hash = "sha256:7b8b4bdc4610193d1e58ebd3fc04be58cd471d7155780eab36a5b3c68d6ec182", size = 196937, upload-time = "2022-09-21T19:43:19.494Z" } +sdist = { url = "https://files.pythonhosted.org/packages/19/6d/253216655a7fe6ffbd605c0b271ac937a6a997e7d2ab6a1c2a0cacd4bd6a/python_irodsclient-3.2.0.tar.gz", hash = "sha256:a7e1c765b5c24c7cc579334e06c32bc76bf4e9ffac3209abcea853f9440ff2e4", size = 308578, upload-time = "2025-08-28T02:07:46.234Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/ad/cd/592c4df3f9772dbb6d8dcbaa388554e21d3db059ad4660bfb3dc705e9dab/python_irodsclient-1.1.5-py2.py3-none-any.whl", hash = "sha256:4665d9ef6837054f3526a8fc183f89cd14a26865d24c38de2a111c6920cdb510", size = 181046, upload-time = "2022-09-21T19:43:16.533Z" }, + { url = "https://files.pythonhosted.org/packages/c8/91/e4c5f29a8021b1d98d32f7b2374251cf8ed3504752f49daff19fe4bac5df/python_irodsclient-3.2.0-py3-none-any.whl", hash = "sha256:a3efa041e771b1e65b69153017d70ba6dcc53e8db279a6ce97dd65b81cf59c3d", size = 279411, upload-time = "2025-08-28T02:07:44.979Z" }, ] [[package]]