diff --git a/docs/source/core.md b/docs/source/core.md index 876cb71..b96d5ec 100644 --- a/docs/source/core.md +++ b/docs/source/core.md @@ -15,6 +15,34 @@ owilix/core/manager/ └── logging.py # DBLog transaction logging ``` +## Architecture + +The core architecture revolves around the **`OWIlixManager`**, which acts as the central hub for: + +1. **Configuration**: Managing `OWIlixConfig` and environment variables. +2. **Session**: Holding the `OWILIXSession` (wrapping `py4lexis` session) for authentication. +3. **Repositories**: Creating and aggregating `AbstractRepository` instances (Local, Lexis, etc.). + +### Component Interaction + +```mermaid +graph TD + CLI[OWILIX CLI] --> Manager[OWIlixManager] + Manager --> Config[OWIlixConfig] + Manager --> Session[OWILIXSession] + Manager --> Remote[AggregatedRepository] + Manager --> Local[LocalRepository] + + Session --> Lexis[LexisSession (py4lexis)] + Remote --> LexisRepo[LexisRepository] + LexisRepo --> FS[Http2IrodsFileSystem] + FS --> Lexis +``` + +### Reference Documentation +- [Repository System](repository.md) +- [FSSpec Integration](fsspec_integration.md) + ## Quick Start ```python @@ -153,5 +181,5 @@ progress.stop() ## See Also -- [FSSPEC Integration](fsspec_integration.md) - +- [Repository Package](repository.md) - Dataset storage abstractions +- [FSSPEC Integration](fsspec_integration.md) - iRODS HTTP filesystem diff --git a/docs/source/fsspec_integration.md b/docs/source/fsspec_integration.md index 39895cc..4cafe90 100644 --- a/docs/source/fsspec_integration.md +++ b/docs/source/fsspec_integration.md @@ -1,7 +1,26 @@ +# iRODS# FSSpec Integration -# iRODS FSSPEC Integration +OWILIX uses the [Filesystem Spec (fsspec)](https://filesystem-spec.readthedocs.io/) library to provide a unified interface for accessing different storage backends. -The `owilix.core.fsspec` module provides a filesystem specification (fsspec) implementation for iRODS, enabling direct interaction with iRODS data objects and collections using standard Python file APIs. This integration is particularly optimized for data science workflows, including direct querying of Parquet files via DuckDB. +## `Http2IrodsFileSystem` + +The core component for LEXIS integration is `Http2IrodsFileSystem`. This custom implementation allows standard filesystem operations (ls, open, get) over the LEXIS iRODS HTTP API. + +**Key Characteristics:** +- **HTTP-based**: Works through firewalls where standard iRODS TCP (1247) is blocked. +- **ReadOnly**: Primarily designed for reading/listing datasets. +- **Session-aware**: Uses `token_provider` callback to maintain valid authentication during long operations. + +### Usage +Typically used internally by `LexisRepository`: +```python +fs = Http2IrodsFileSystem( + irods_http_url="https://..../irods/api/v1", + token_provider=session.get_token +) +fs.ls("/zone/project/...") +``` +(fsspec) implementation for iRODS, enabling direct interaction with iRODS data objects and collections using standard Python file APIs. This integration is particularly optimized for data science workflows, including direct querying of Parquet files via DuckDB. ## Overview diff --git a/docs/source/repository.md b/docs/source/repository.md new file mode 100644 index 0000000..3680928 --- /dev/null +++ b/docs/source/repository.md @@ -0,0 +1,157 @@ +# OWILIX Repository Package + +The `owilix.core.repository` package provides abstractions for local and remote dataset storage. + +**See also:** [Core Manager](core.md) | [FSSPEC Integration](fsspec_integration.md) + +## Package Structure + +``` +owilix/core/repository/ +├── __init__.py # Package exports +├── base.py # AbstractRepository, PerformanceMetric +├── file.py # FileBasedRepository, LocalRepository +├── lexis.py # LexisRepository (py4lexis 5.x HTTP) +└── aggregate.py # AggregatedRepository +``` + +## Quick Start + +```python +from owilix.core.repository import LexisRepository, LocalRepository, AggregatedRepository + +# Via manager (recommended) +from owilix.core import OWILIXManager +manager = OWILIXManager() +repos = manager.remote_data # AggregatedRepository + +# Direct instantiation +lexis = LexisRepository(manager, lexis_project_id="my-project") +local = LocalRepository(manager, path="~/.owi/{access}", repo_name="local") +``` + +## Repository Types + +### LexisRepository + +LEXIS HTTP repository using py4lexis 5.x session - **no python-irodsclient required**. + +```python +from owilix.core.repository import LexisRepository + +repo = LexisRepository( + manager=manager, + lexis_project_id="proj862c5962623246664c1fda27b7afb108", # Optional + repo_name="lexis" +) + +# List datasets +datasets = repo.list(access="public") + +# Get files +files = repo.files(dataset, files_glob="**/*.parquet") + +# Download +repo.get(dataset, "data.parquet", "/local/path") +``` + +**Key features:** +- Uses `OWILexisDatasetAPI` for DDI metadata +- Uses `Http2IrodsFileSystem` for file operations +- Gets URL from `session.irods_http_api_url` (no manual config) + +### FileBasedRepository + +Filesystem-based repository using fsspec. + +```python +from owilix.core.repository import FileBasedRepository + +repo = FileBasedRepository( + manager=manager, + path="~/.owi/{access}", # Must contain {access} + protocol="file", # or "s3" + repo_name="local" +) +``` + +### LocalRepository + +Alias for `FileBasedRepository` with `protocol="file"`. + +### AggregatedRepository + +Aggregates multiple repositories with performance-based selection. + +```python +from owilix.core.repository import AggregatedRepository + +agg = AggregatedRepository({ + "it4i": lexis_repo, + "local": local_repo +}) + +# Lists from all repos, deduplicates by best performance +datasets = agg.list(access="public") + +# Performance stats +agg.store_stats_to_file("stats.json") +agg.get_performance_stats() +``` + +## Performance Tracking + +All repositories track list, upload, and download performance. + +```python +from owilix.core.repository import RepoPerformanceStats, PerformanceMetric + +# Access stats +stats = repo.performance_stats +print(stats.list_response_time.get_overall_average()) +print(stats.download_bandwidth.get_overall_average()) + +# Serialize +data = stats.to_json() +restored = RepoPerformanceStats.from_json(data) +``` + +## Abstract Base Class + +```{eval-rst} +.. autoclass:: owilix.core.repository.AbstractRepository + :members: list, files, files_details, exists, put, get, rm, create, status +``` + +## Configuration + +Repositories are configured in `~/.owi/owilix.cfg`: + +```yaml +repositories: + config: + it4i: + repository: lexis + options: + project_name: openwebsearch + lexis_project_id: proj862c5962623246664c1fda27b7afb108 + local: + repository: file + options: + path: ~/.owi/{access} + selected_remote: + - it4i + selected_local: + - local +``` + +**Repository types:** +- `lexis+http`: LexisRepository (HTTP API) +- `file`: LocalRepository +- `s3a`: S3FileBasedRepository + +## Legacy Aliases + +For backward compatibility: +- `LEXISIrodsRepository` → `LexisRepository` +- `LEXISIrodsHTTPRepository` → `LexisRepository` diff --git a/owilix/core/__init__.py b/owilix/core/__init__.py index 80c3e02..43ff0c9 100644 --- a/owilix/core/__init__.py +++ b/owilix/core/__init__.py @@ -1,3 +1,3 @@ from owilix.core.manager import OWIlixManager, load_and_check, _targetProjectHash -from .repository import LocalRepository, LEXISIrodsRepository, AggregatedRepository +from owilix.core.repository import LocalRepository, AggregatedRepository, LexisRepository from .metadata import infer_metadata_from_files \ No newline at end of file diff --git a/owilix/core/fsspec/http2irods.py b/owilix/core/fsspec/http2irods.py index 7192eaf..7ebe41d 100644 --- a/owilix/core/fsspec/http2irods.py +++ b/owilix/core/fsspec/http2irods.py @@ -30,6 +30,7 @@ from fsspec.spec import AbstractBufferedFile from irods_http_client.collection_operations import Collections from irods_http_client.data_object_operations import DataObjects +from irods_http_client.models.collection import iRODSCollecion logger = logging.getLogger("owilix.fsspec") @@ -143,41 +144,61 @@ class Http2IrodsFileSystem(fsspec.AbstractFileSystem): """ List contents of a collection (directory). + Uses internal `iRODSCollection` wrapper and GenQuery for reliable retrieval of + both subcollections and data objects. + Args: - path: iRODS collection path - detail: If True, return list of dicts with metadata + path: iRODS collection path (absolute or relative) + detail: If True, return list of dicts with metadata (name, type, size, mtime) Returns: - List of paths or list of dicts with file info + List of absolute paths (str) if detail=False, or list of dicts if detail=True. """ try: - # Use Collections.list() which returns entries - result = self.collections.list(path) + path = self._strip_protocol(path) + # Ensure path starts with / + if not path.startswith("/"): + path = "/" + path - # Handle nested response structure - if isinstance(result, dict): - data = result.get("data", result) - entries = data.get("entries", []) - else: - entries = [] + logger.debug(f"Listing path: {path}") + # Use iRODSCollecion wrapper which uses GenQuery + # This is more reliable than op=list + coll = iRODSCollecion(self.collections) + try: + coll.initialize(path) + except Exception as e: + # If collection doesn't exist or error init + logger.debug(f"Collection init failed for {path}: {e}") + raise FileNotFoundError(path) + items = [] - for entry in entries: - entry_path = entry if isinstance(entry, str) else str(entry) + + # Get subcollections + for sub in coll.subcollections: if detail: - # Get info for each entry to determine type - try: - info = self.info(entry_path) - items.append(info) - except Exception: - items.append({ - "name": entry_path, - "type": "unknown", - "size": 0, - }) + items.append({ + "name": sub.path, + "type": "directory", + "size": 0, + "mtime": sub.modify_time # Might be int or str + }) + else: + items.append(sub.path) + + # Get data objects + for obj in coll.data_objects: + if detail: + items.append({ + "name": obj.path, + "type": "file", + "size": obj.size, + "mtime": obj.modify_time + }) else: - items.append(entry_path) + items.append(obj.path) + logger.debug(f"Found {len(items)} items in {path}") return items except Exception as e: diff --git a/owilix/core/manager/manager.py b/owilix/core/manager/manager.py index 9fd72b9..ee79667 100644 --- a/owilix/core/manager/manager.py +++ b/owilix/core/manager/manager.py @@ -182,6 +182,43 @@ class OWIlixConfig: self.config["general"]["license"] = license_info["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. + + Tests if the server is online with a HEAD request first, + 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() + except requests.RequestException as e: + logger.warning(f"Server is offline or not responding: {e}") + return {} + + # 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: + logger.warning(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]: """Fetch and merge server configuration.""" @@ -273,18 +310,18 @@ class OWIlixManager: """ Central orchestrator for OWILIX operations. - Manages: - - Configuration loading and persistence - - Session authentication (via OWILIXSession) - - Repository instantiation and aggregation - - Logging and statistics + This class serves as the main entry point for the application logic, managing: + - **Configuration**: Loading, validation, and persistence via `OWIlixConfig`. + - **Authentication**: Unified session management using `OWILIXSession` (wrapping `LexisSession`). + - **Repositories**: Instantiation and aggregation of configured repositories (Local, Remote, LEXIS). + - **Logging & Metrics**: Centralized event logging and performance tracking. 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. + name (str): Project name. + config (OWIlixConfig): The active configuration object. + session (OWILIXSession): The authenticated session wrapper. + local (LocalRepository): The primary local repository instance. + remote_data (AggregatedRepository): The aggregated view of all remote repositories. """ def __init__(self, @@ -440,7 +477,7 @@ class OWIlixManager: if _login.get("username"): self.logger.info("Using credentials from configuration") return LexisSession( - suppress_print=True, in_cli=True, + suppress_print=True, login_method="credentials", log_file=self.get_lexis_log_filename(), username=_login.get("username"), @@ -454,9 +491,9 @@ class OWIlixManager: try: if _refresh_token: session = LexisSession( - suppress_print=True, in_cli=True, + suppress_print=True, offline_access=True, - login_method="offline", + login_method="token", log_file=self.get_lexis_log_filename(), refresh_token=_refresh_token ) @@ -467,14 +504,14 @@ class OWIlixManager: except Py4LexisAuthException: # Fallback to URL login session = LexisSession( - suppress_print=False, in_cli=True, + suppress_print=False, 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() + _new_token = session.get_refresh_token() if _new_token and _new_token != _refresh_token: self._refresh_token_io(_new_token) @@ -521,6 +558,104 @@ class OWIlixManager: """Persist repository performance statistics.""" self._remote_data.store_stats_to_file(self._repo_stats_file) + def get_local_collection_path(self, access: str, collection: str) -> str: + """Get local path for a collection.""" + from owilix.core.utils import check_access + check_access(access) + return os.path.join(self.owi_path, access, collection) + + def stats(self, day=None, duration: int = 0, data_center=None, query=None): + """ + Generates statistics about datasets based on specified parameters. + + Args: + day: Specific day to analyze or None for all. + duration: Number of days to include from the specified day. + data_center: Data center name for filtering. + query: Additional query parameters. + + Returns: + pd.DataFrame: Summary DataFrame with statistics. + """ + df, _files = self.remote_data.ls_http(day, duration, data_center, query) + 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]] + 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 parse_specifier(self, specifier: str) -> dict: + """ + Parses a specifier string to extract data query parameters. + + Args: + specifier: Format `:#/=`. + + Returns: + dict: Parsed specifier with data_center, day, duration, query keys. + """ + import re + + result = { + 'data_center': None, + 'day': None, + 'duration': None, + 'query': {} + } + + parts = specifier.split('/') + main_part = parts[0] + filter_parts = parts[1:] if len(parts) > 1 else [] + + main_regex = r'^(?P[\w-]+)(?::(?P[\d-]+|latest)(?:#(?P\d+))?)?$' + match = re.match(main_regex, main_part) + if match: + result.update(match.groupdict()) + + if result['day']: + if result['day'] != 'latest': + try: + result['day'] = dateutil.parser.parse(result['day']) + except ValueError: + result['day'] = None + + if result['duration']: + result['duration'] = int(result['duration']) + + if result.get('data_center') == "all": + result['data_center'] = None + + for filter_part in filter_parts: + for kv in filter_part.split(';'): + if '=' in kv: + key, value = kv.split('=', 1) + result['query'][key] = value + + return result + + +def _flatten_and_count(df, column): + """Flatten a column [a,b,c] and count the occurrences 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() + return df[column].explode().value_counts().reset_index() + else: + return df[column].value_counts().reset_index() + # Backward compatibility alias OWILIXManager = OWIlixManager diff --git a/owilix/core/repository/__init__.py b/owilix/core/repository/__init__.py new file mode 100644 index 0000000..ba61275 --- /dev/null +++ b/owilix/core/repository/__init__.py @@ -0,0 +1,57 @@ +""" + RepoPerformanceStats: Performance metrics for a repository +""" + +from owilix.core.repository.base import ( + AbstractRepository, + PerformanceMetric, + RepoPerformanceStats, + measure_performance, +) + +# Lazy imports to avoid circular dependencies +def __getattr__(name): + if name == "FileBasedRepository": + from owilix.core.repository.file import FileBasedRepository + return FileBasedRepository + elif name == "LocalRepository": + from owilix.core.repository.file import LocalRepository + return LocalRepository + elif name == "S3FileBasedRepository": + from owilix.core.repository.file import S3FileBasedRepository + return S3FileBasedRepository + elif name == "LexisRepository": + from owilix.core.repository.lexis import LexisRepository + return LexisRepository + elif name == "AggregatedRepository": + from owilix.core.repository.aggregate import AggregatedRepository + return AggregatedRepository + # Legacy names for backward compatibility + elif name == "LEXISIrodsRepository": + # Map to LexisRepository for backward compat + from owilix.core.repository.lexis import LexisRepository + return LexisRepository + elif name == "LEXISIrodsHTTPRepository": + from owilix.core.repository.lexis import LexisRepository + return LexisRepository + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + + +__all__ = [ + # Base + "AbstractRepository", + "PerformanceMetric", + "RepoPerformanceStats", + "measure_performance", + # File-based + "FileBasedRepository", + "LocalRepository", + "S3FileBasedRepository", + # LEXIS + "LexisRepository", + # Legacy aliases + "LEXISIrodsRepository", + "LEXISIrodsHTTPRepository", + # Aggregate + "AggregatedRepository", +] diff --git a/owilix/core/repository/aggregate.py b/owilix/core/repository/aggregate.py new file mode 100644 index 0000000..4741437 --- /dev/null +++ b/owilix/core/repository/aggregate.py @@ -0,0 +1,244 @@ +""" +Aggregated Repository + +Aggregates multiple repositories under one API with performance-based selection. +""" + +import json +import logging +import os +import time +from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import datetime +from typing import Dict, List, Union, Sequence + +import fsspec + +from owilix.core.repository.base import ( + AbstractRepository, + RepoPerformanceStats, + measure_performance, +) +from owilix.core.metadata import Dataset + +logger = logging.getLogger("owilix") + + +class AggregatedRepository: + """ + Aggregates multiple repositories under one API. + + For each dataset (identified by d.metadata["id"]), if duplicates exist, + the repository with the best performance is chosen. + """ + + def __init__( + self, + repositories: Dict[str, AbstractRepository], + weight_response_time: float = 1.0, + weight_bandwidth: float = 1.0 + ): + self.repositories = repositories + self.weight_response_time = weight_response_time + self.weight_bandwidth = weight_bandwidth + + @staticmethod + def single_repo_from_url(url: str) -> AbstractRepository: + """Create a single repository from a URL.""" + import urllib.parse + from owilix.core.repository.file import FileBasedRepository + + parsed = urllib.parse.urlparse(url) + if parsed.scheme == "file": + return FileBasedRepository(None, parsed.path) + elif parsed.scheme in ("s3", "s3a"): + return FileBasedRepository(None, parsed.path, protocol="s3") + else: + raise NotImplementedError(f"Repository for scheme {parsed.scheme} not implemented") + + def status(self) -> Dict[str, dict]: + return {k: r.status() for k, r in self.repositories.items()} + + def get_repo_names(self) -> set: + return set(self.repositories.keys()) + + def get_repos(self, repo_key: Union[str, List[str], None] = None): + if repo_key is None: + return self.repositories + elif isinstance(repo_key, list): + return {k: r for k, r in self.repositories.items() if k in repo_key} + else: + return {k: r for k, r in self.repositories.items() if k == repo_key} + + def get_single_repo(self, repo_key: str): + _repos = self.get_repos(repo_key) + if len(_repos) == 0: + return None + if len(_repos) != 1: + raise ValueError(f"Datacenter {repo_key} not found or multiple found: {list(_repos.keys())}") + return list(_repos.values())[0] + + def get_repo_for_zone(self, zone: str): + for _, r in self.repositories.items(): + if getattr(r, "zone", None) == zone: + return r + return None + + def exists_repo(self, repo_key: str) -> bool: + return repo_key in self.repositories + + def _compute_repo_score(self, repo: AbstractRepository) -> float: + """Compute score for a repository based on performance stats.""" + avg_response_time = repo.performance_stats.list_response_time.get_overall_average() + avg_bandwidth = repo.performance_stats.list_bandwidth.get_overall_average() + score = self.weight_response_time * (1.0 / avg_response_time if avg_response_time > 0 else 0) + score += self.weight_bandwidth * avg_bandwidth + return score + + def _list_repo_wrapper(self, repo: AbstractRepository, access: str, + day: Union[str, datetime, None], duration: int, + query, cb_progress): + start = time.time() + result = repo.list(access, day, duration, query, cb_progress) + elapsed_time = time.time() - start + return result, elapsed_time + + def list( + self, + datacenter: Union[str, List[str], None] = None, + access: str = "public", + day: Union[str, datetime, None] = None, + duration: int = 0, + query=None, + cb_progress=None, + ignore_data_centers: List[str] = None, + ) -> list: + """List datasets from all or specified repositories.""" + results = [] + repos = self.get_repos(datacenter).items() + if ignore_data_centers is not None: + repos = [(k, v) for k, v in repos if k not in ignore_data_centers] + + with ThreadPoolExecutor() as executor: + future_to_dc = { + executor.submit(self._list_repo_wrapper, r, access, day, duration, query, cb_progress): dc + for dc, r in repos + } + for future in as_completed(future_to_dc): + dc = future_to_dc[future] + try: + repo_result, elapsed_time = future.result() + repo = self.repositories[dc] + repo.update_performance_stats("list", elapsed_time, len(repo_result)) + results.extend(repo_result) + except Exception as e: + logger.exception(f"Could not list datasets in datacenter {dc}: {e}") + + # Deduplicate by dataset ID, keeping best performing repo + grouped: Dict[str, List[Dataset]] = {} + for ds in results: + dataset_id = ds.metadata.get("id") + grouped.setdefault(dataset_id, []).append(ds) + + best_results = [] + for dataset_id, ds_list in grouped.items(): + best_ds = max(ds_list, key=lambda ds: self._compute_repo_score(ds.repository)) + best_results.append(best_ds) + return best_results + + def exists(self, dataset: Dataset) -> bool: + repo = self.get_single_repo(dataset.repository) + if repo is None: + return False + return repo.exists(dataset) + + def files(self, dataset: Dataset, files_glob: str | Sequence[str] = None) -> list: + return dataset.repository.files(dataset, files_glob) + + def files_details(self, dataset: Dataset, files_glob: str | Sequence[str] = None, count_rows=False) -> list: + return dataset.repository.files_details(dataset, files_glob, count_rows=count_rows) + + def create( + self, + datacenter: str, + access: str = "public", + collectionName: str = "main", + metadata=None + ) -> Dataset: + repo = self.get_single_repo(datacenter) + if repo is None: + return None + return repo.create(access, collectionName, metadata or {}) + + def delete(self, dataset: Dataset): + return dataset.repository.delete(dataset) + + @measure_performance("upload", lambda self, dataset, local_path, local_file, filesystem=None: local_path) + def put(self, dataset: Dataset, local_path: str, local_file: str, filesystem: fsspec.AbstractFileSystem = None) -> str: + if filesystem is not None: + raise NotImplementedError("Putting file to other filesystem not implemented yet.") + return dataset.repository.put(dataset, local_path, local_file) + + @measure_performance("download", lambda self, dataset, file, local_path, filesystem=None: os.path.join(local_path, os.path.basename(file))) + def get(self, dataset: Dataset, file: str, local_path: str, filesystem: fsspec.AbstractFileSystem = None): + repo = dataset.repository + start = time.time() + result = repo.get(dataset, file, local_path, filesystem) + elapsed = time.time() - start + try: + local_file = os.path.join(local_path, os.path.basename(file)) + file_size = os.path.getsize(local_file) + except Exception: + file_size = 0 + repo.update_performance_stats("download", elapsed, file_size) + return result + + def rm(self, dataset: Dataset, files: List[str]): + return dataset.repository.rm(dataset, files) + + def checks( + self, + datacenter: Union[str, List[str], None] = None, + access: str = "public", + day: Union[str, datetime, None] = None, + duration: int = 0, + query=None, + cb_progress=None + ) -> list: + results = [] + for dc, r in self.get_repos(datacenter).items(): + try: + results.extend(r.checks(access, day, duration, query, cb_progress)) + except Exception as e: + logger.exception(f"Checks failed on datacenter {dc}: {e}") + return results + + def update_metadata(self, dataset: Dataset, force: bool = False): + return dataset.repository.update_metadata(dataset, force) + + def change_id(self, dataset: Dataset, new_id: str): + return dataset.repository.change_id(dataset, new_id) + + def __str__(self): + repo_list = ", ".join(self.repositories.keys()) + return f"AggregatedRepository({repo_list})" + + def store_stats_to_file(self, filename: str): + """Store performance statistics to a JSON file.""" + stats = {key: repo.performance_stats.to_json() for key, repo in self.repositories.items()} + with open(filename, "w") as f: + json.dump(stats, f) + + def load_stats_from_file(self, filename: str): + """Load performance statistics from a JSON file.""" + if os.path.exists(filename): + with open(filename, "r") as f: + stats = json.load(f) + else: + stats = {} + for key, stat in stats.items(): + if key in self.repositories: + self.repositories[key].performance_stats = RepoPerformanceStats.from_json(stat) + + def get_performance_stats(self): + return {k: r.performance_stats.to_json() for k, r in self.repositories.items()} diff --git a/owilix/core/repository/base.py b/owilix/core/repository/base.py new file mode 100644 index 0000000..995ace8 --- /dev/null +++ b/owilix/core/repository/base.py @@ -0,0 +1,332 @@ +""" +Repository Base Classes + +This module provides the abstract base classes and performance statistics +for all repository implementations. + +Classes: + PerformanceMetric: Maintains running statistics for a single indicator + RepoPerformanceStats: Holds performance metrics for a repository + AbstractRepository: Abstract base class for all repositories +""" + +import logging +import os +from abc import abstractmethod +from datetime import datetime +from typing import List, Union, Sequence + +import fsspec + +from owilix.core.metadata import Dataset + +logger = logging.getLogger("owilix") + + +############################################################################### +# Performance Stats Classes +############################################################################### + +class PerformanceMetric: + """ + Maintains running statistics for a single performance indicator. + + Uses exponentially moving average (EMA) and a buffer of recent values. + """ + def __init__(self, alpha: float = 0.5, buffer_size: int = 20, weight_ema: float = 0.5): + self.alpha = alpha + self.buffer_size = buffer_size + self.weight_ema = weight_ema + self.ema = None + self.buffer = [] + self.count = 0 + + def update(self, value: float): + """Updates the metric with a new value.""" + self.count += 1 + if self.ema is None: + self.ema = value + else: + self.ema = self.alpha * value + (1 - self.alpha) * self.ema + self.buffer.append(value) + if len(self.buffer) > self.buffer_size: + self.buffer.pop(0) + + def get_buffer_average(self) -> float: + """Returns the average of the buffered values.""" + if not self.buffer: + return 0.0 + return sum(self.buffer) / len(self.buffer) + + def get_overall_average(self) -> float: + """Returns weighted blend of EMA and buffer average.""" + buffer_avg = self.get_buffer_average() + ema = self.ema if self.ema is not None else 0.0 + return self.weight_ema * ema + (1 - self.weight_ema) * buffer_avg + + def to_json(self) -> dict: + """Serializes the metric to a JSON-serializable dict.""" + return { + "alpha": self.alpha, + "buffer_size": self.buffer_size, + "weight_ema": self.weight_ema, + "ema": self.ema, + "buffer": self.buffer, + "count": self.count + } + + @classmethod + def from_json(cls, data: dict) -> "PerformanceMetric": + """Creates a PerformanceMetric instance from a dict.""" + obj = cls(alpha=data.get("alpha", 0.5), + buffer_size=data.get("buffer_size", 20), + weight_ema=data.get("weight_ema", 0.5)) + obj.ema = data.get("ema") + obj.buffer = data.get("buffer", []) + obj.count = data.get("count", 0) + return obj + + +class RepoPerformanceStats: + """ + Holds performance metrics for a repository. + + Tracks: list_response_time, list_bandwidth, upload_bandwidth, download_bandwidth + """ + def __init__(self, alpha: float = 0.5, buffer_size: int = 20, weight_ema: float = 0.5): + self.list_response_time = PerformanceMetric(alpha, buffer_size, weight_ema) + self.list_bandwidth = PerformanceMetric(alpha, buffer_size, weight_ema) + self.upload_bandwidth = PerformanceMetric(alpha, buffer_size, weight_ema) + self.download_bandwidth = PerformanceMetric(alpha, buffer_size, weight_ema) + + def to_json(self) -> dict: + """Serializes the repository performance stats to a JSON-serializable dict.""" + return { + "list_response_time": self.list_response_time.to_json(), + "list_bandwidth": self.list_bandwidth.to_json(), + "upload_bandwidth": self.upload_bandwidth.to_json(), + "download_bandwidth": self.download_bandwidth.to_json() + } + + @classmethod + def from_json(cls, data: dict) -> "RepoPerformanceStats": + """Creates a RepoPerformanceStats instance from a dict.""" + obj = cls() + obj.list_response_time = PerformanceMetric.from_json(data.get("list_response_time", {})) + obj.list_bandwidth = PerformanceMetric.from_json(data.get("list_bandwidth", {})) + obj.upload_bandwidth = PerformanceMetric.from_json(data.get("upload_bandwidth", {})) + obj.download_bandwidth = PerformanceMetric.from_json(data.get("download_bandwidth", {})) + return obj + + +############################################################################### +# Abstract Repository +############################################################################### + +class AbstractRepository(ABC): + """ + Abstract base class for all OWILIX repositories. + + Defines the standard interface for dataset enumeration (`ls`), file listing (`files`), + and dataset retrieval (`pull`). + + Attributes: + commands: Dictionary of registered commands for this repository type. + _metrics: Dictionary storing `PerformanceMetric` objects for operation monitoring. + """ + _filesystem = None + _repo_name = "unnamed_repo" + _async_capable = False + _tags = [] + _description = "Abstract Repository" + _backend_name = "unknown" + + def __init__(self): + self.performance_stats = RepoPerformanceStats() + + @property + def tags(self): + return self._tags + + @property + def description(self): + return self._description + + @property + def fs(self): + return self._filesystem + + @property + def pathsep(self): + return "/" if self.fs.protocol != "file" else os.pathsep + + @property + def repo_name(self): + return self._repo_name + + def _add_details_to_status(self, _d): + _d.update({ + "description": self.description, + "tags": self.tags, + "backend": self._backend_name + }) + return _d + + def status(self) -> dict: + return self._add_details_to_status({ + "status": False, + "message": "not implemented", + "public": False, "user": False, "project": False + }) + + def update_performance_stats(self, op: str, elapsed_time: float, data_size: float): + """ + Updates performance statistics for a given operation. + + Parameters + ---------- + op : str + Operation type ("list", "upload", or "download"). + elapsed_time : float + Time in seconds taken by the operation. + data_size : float + For list, number of items; for upload/download, bytes transferred. + """ + if op == "list": + self.performance_stats.list_response_time.update(elapsed_time) + new_bw = (data_size / elapsed_time) if elapsed_time > 0 else 0 + self.performance_stats.list_bandwidth.update(new_bw) + elif op == "upload": + new_bw = (data_size / elapsed_time) if elapsed_time > 0 else 0 + self.performance_stats.upload_bandwidth.update(new_bw) + elif op == "download": + new_bw = (data_size / elapsed_time) if elapsed_time > 0 else 0 + self.performance_stats.download_bandwidth.update(new_bw) + + @abstractmethod + def list( + self, + access: str = "public", + day: Union[str, datetime, None] = None, + duration: int = 0, + query=None, + cb_progress=None + ) -> List[Dataset]: + raise NotImplementedError() + + def is_async_capable(self): + return self._async_capable + + def get_async_loop(self): + return None + + @abstractmethod + def exists(self, dataset: Dataset) -> bool: + raise NotImplementedError() + + @abstractmethod + def files_details( + self, dataset: Dataset, files_glob: str | Sequence[str] = None, count_rows: bool = False + ) -> list: + raise NotImplementedError() + + @abstractmethod + def readlines(self, dataset: Dataset, file_name: str) -> Union[str, None]: + raise NotImplementedError() + + @abstractmethod + def writelines(self, dataset: Dataset, file_name: str, content: str) -> None: + raise NotImplementedError() + + @abstractmethod + def files(self, dataset: Dataset, files_glob: str | Sequence[str] = None) -> list: + raise NotImplementedError() + + @abstractmethod + def delete(self, dataset: Dataset) -> None: + raise NotImplementedError() + + @abstractmethod + def delete_by_id(self, id: str) -> None: + raise NotImplementedError() + + @abstractmethod + def put( + self, dataset: Dataset, local_path: str, rel_filename: str, + filesystem: fsspec.AbstractFileSystem = None + ) -> str: + raise NotImplementedError() + + @abstractmethod + def get( + self, dataset: Dataset, file: str, local_path: str, + filesystem: fsspec.AbstractFileSystem = None + ) -> None: + raise NotImplementedError() + + @abstractmethod + def rm(self, dataset: Dataset, files: List[str]) -> None: + raise NotImplementedError() + + def checks( + self, + access: str = "public", + day: Union[str, datetime, None] = None, + duration: int = 0, + query=None, + cb_progress=None + ) -> list: + return [] + + def update_metadata(self, dataset: Dataset, force: bool = False) -> list: + return [] + + @abstractmethod + def create( + self, + access: str = "public", + collectionName: str = "main", + metadata={} + ) -> Dataset: + raise NotImplementedError() + + @abstractmethod + def change_id(self, dataset: Dataset, new_id: str): + pass + + def __str__(self): + return f"{self.repo_name} {super().__str__()}" + + +############################################################################### +# Utility Functions +############################################################################### + +def measure_performance(operation: str, file_path_getter): + """ + Decorator to measure elapsed time and file size for performance stats. + + Args: + operation (str): The operation type ('upload' or 'download'). + file_path_getter (Callable): Function to extract file path from args. + """ + import functools + import time + + def decorator(func): + @functools.wraps(func) + def wrapper(self, *args, **kwargs): + start_time = time.time() + result = func(self, *args, **kwargs) + elapsed_time = time.time() - start_time + + try: + file_path = file_path_getter(self, *args, **kwargs) + file_size = os.path.getsize(file_path) if os.path.exists(file_path) else 0 + self.update_performance_stats(operation, elapsed_time, file_size) + except Exception: + pass + + return result + return wrapper + return decorator diff --git a/owilix/core/repository/file.py b/owilix/core/repository/file.py new file mode 100644 index 0000000..aa2db63 --- /dev/null +++ b/owilix/core/repository/file.py @@ -0,0 +1,436 @@ +""" +File-based Repository Implementations + +Provides repositories for local and remote filesystems using fsspec. + +Classes: + FileBasedRepository: Base filesystem repository + LocalRepository: Local filesystem repository + S3FileBasedRepository: S3-based repository +""" + +import asyncio +import json +import logging +import os +import re +import fnmatch +import uuid +from datetime import datetime +from typing import List, Union, Sequence + +import fsspec +from fsspec.callbacks import Callback + +from owilix.core.repository.base import ( + AbstractRepository, + measure_performance, +) +from owilix.core.metadata import Dataset +from owilix.core.utils import check_path_for_uuid_filename, fill_file_details + +logger = logging.getLogger("owilix") + + +def _normalize_globs(files_glob: str | Sequence[str] | None) -> List[str] | None: + if files_glob is None: + return None + if isinstance(files_glob, str): + return [files_glob] + return [g for g in files_glob if g] + + +def _compile_globs(globs: List[str]) -> list[re.Pattern]: + return [re.compile(fnmatch.translate(g)) for g in globs] + + +def _matches_any(path: str, relpath: str, patterns: list[re.Pattern]) -> bool: + for rx in patterns: + if rx.fullmatch(relpath) or rx.fullmatch(path): + return True + return False + + +class DotCallback(Callback): + """Progress callback that prints dots.""" + def __init__(self): + super().__init__() + self.bytes_transferred = 0 + + def __call__(self, bytes): + self.bytes_transferred += bytes + if self.bytes_transferred >= 1_000_000: # 1 MB + print('.', end='', flush=True) + self.bytes_transferred = 0 + + +class FileBasedRepository(AbstractRepository): + """ + A repository implementation for local or remote filesystems using fsspec. + Requires the base path to contain '{access}'. + """ + + @staticmethod + def get_fsspec_from_config(url, protocol="file", **kwargs): + _async = kwargs.get("async", False) + if protocol == "file": + return fsspec.filesystem(protocol), os.path.expanduser(url), fsspec.filesystem(protocol, synchronous=_async) if _async else None + elif protocol.startswith("s3"): + import s3fs + s3fs.S3FileSystem.fsid = property(lambda self: "s3a" + self.client_kwargs.get("endpoint", "")) + fs = fsspec.filesystem( + protocol, + key=kwargs.get("key", None), + secret=kwargs.get("secret", None), + anon=kwargs.get("anonymous", False), + client_kwargs={"endpoint_url": kwargs.get("endpoint", None)} + ) + fs_async = fsspec.filesystem(protocol, key=kwargs.get("key", None), + secret=kwargs.get("secret", None), + client_kwargs={"endpoint_url": kwargs.get("endpoint", None)}, + synchronous=not _async) if _async else None + return fs, url, fs_async + else: + return fsspec.filesystem(protocol), url, fsspec.filesystem(protocol, asynchronous=_async) if _async else None + + def __init__( + self, + manager, + path: str, + repo_name: str = None, + protocol: str = "file", + tags: List[str] = None, + description: str = "File-based repository", + **kwargs + ): + super().__init__() + self.manager = manager + if "{access}" not in path: + raise ValueError("The path must contain the access level as {access}") + self._filesystem, self.path, self._filesystem_async = FileBasedRepository.get_fsspec_from_config(path, protocol, **kwargs) + self._repo_name = repo_name if repo_name else "unnamed_repo" + self._collection_name_in_path = kwargs.get("collections_in_path", True) + self._retry_count = kwargs.get("retry_count", 3) + self._chunk_size = kwargs.get("chunk_size", 5 * 1024 * 1024) + self._tags = tags if tags is not None else ["main"] + self._description = description + self._backend_name = "filesystem" + + def __str__(self): + return f"{self.__class__.__name__}@{self._filesystem}/{self.path}<{'async+' if self._filesystem_async is not None else 'async-'}>" + + def status(self) -> dict: + try: + public_exists = any(self.fs.exists(p) for p in self._get_collection_paths("public")) + project_exists = any(self.fs.exists(p) for p in self._get_collection_paths("project")) + user_exists = any(self.fs.exists(p) for p in self._get_collection_paths("user")) + return self._add_details_to_status({ + "status": True, "message": "connected", + "public": public_exists, "user": user_exists, "project": project_exists + }) + except Exception as e: + return self._add_details_to_status({ + "status": False, + "message": f"Repository or path not available: {str(e)}", + "public": False, "user": False, "project": False + }) + + def exists(self, dataset: Dataset) -> bool: + return self.fs.exists(self._get_path(dataset)) + + def is_async_capable(self): + self._async_capable = (self._filesystem_async is not None and + hasattr(self._filesystem_async, "async_impl") and + getattr(self._filesystem_async, "async_impl")) + return self._async_capable + + def get_async_loop(self): + if self.is_async_capable(): + return self._filesystem_async.get_loop() if self._filesystem_async is not None else None + return None + + def list( + self, + access: str = "public", + day: Union[str, datetime, None] = None, + duration: int = 0, + query=None, + cb_progress=None + ) -> list: + collection_paths = self._get_collection_paths(access) + _dirs = [p for p in collection_paths if self.fs.exists(p)] + if self._collection_name_in_path: + _dirs = [e["name"] for p in _dirs for e in self.fs.listdir(p) if e["type"] == "directory"] + json_files = [] + for subdir in _dirs: + if not self.fs.exists(subdir): + continue + candidates = self.fs.ls(subdir) + for c in candidates: + if c.endswith(".json"): + json_files.append(c) + datasets = [] + for file_path in json_files: + dataset_dir = file_path.replace(".json", "") + if not check_path_for_uuid_filename(os.path.basename(dataset_dir)): + continue + data = self.load_metadata(file_path) + data["access"] = access + data.pop("path", None) + ds = Dataset(repository=self, path=dataset_dir, **data) + datasets.append(ds) + return Dataset.filter_datasets(datasets, day, duration, query) + + def files_details( + self, + dataset: Dataset, + files_glob: str | Sequence[str] | None = None, + count_rows: bool = False, + ) -> list: + """Return details for files matching one or more glob patterns.""" + file_list = self.files(dataset, files_glob) + root_dir = self._get_path(dataset) + return fill_file_details(self.fs, file_list, root=root_dir, do_count=count_rows) + + def files( + self, + dataset: Dataset, + files_glob: str | Sequence[str] | None = None, + ) -> Sequence[str]: + """Return a list of files under the dataset path matching glob patterns.""" + dataset_path = self._get_path(dataset) + + if files_glob is None: + patterns: Sequence[str] = ["**/*"] + elif isinstance(files_glob, str): + patterns = [files_glob] + else: + patterns = list(files_glob) + + def _is_absolute_like(p: str) -> bool: + return ("://" in p or os.path.isabs(p) or p.startswith(dataset_path)) + + hits: set[str] = set() + + for pat in patterns: + full_pattern = pat if _is_absolute_like(pat) else os.path.join(dataset_path, pat) + result = self.fs.glob(full_pattern, detail=True) + if isinstance(result, dict): + for path, info in result.items(): + if info.get("type") in (None, "file"): + hits.add(path) + else: + for path in result: + if not self.fs.isdir(path): + hits.add(path) + + return sorted(hits) + + def _get_collection_paths(self, access: str) -> List[str]: + formatted_path = self.path.format(access=access) + return [os.path.join(formatted_path, "")] + + def _get_path(self, dataset: Dataset) -> str: + slice_name = getattr(dataset, "collectionName", "main") + return os.path.join(self.path.format(access=dataset.access), slice_name, dataset.metadata["id"]) + + def create( + self, + access: str = "public", + collectionName: str = "main", + metadata=None + ) -> Dataset: + metadata = metadata or {} + metadata.pop("collectionName", None) + metadata.pop("access", None) + ds = Dataset(repository=self, path=None, collectionName=collectionName, access=access, **metadata) + if not ds.metadata.get("id"): + ds.metadata["id"] = str(uuid.uuid4()) + dataset_path = self._get_path(ds) + ds.path = dataset_path + if self.fs.exists(dataset_path): + raise ValueError(f"Dataset with ID {ds.metadata['id']} already exists at {dataset_path}") + error = self.update_metadata(ds) + if error: + raise ValueError(f"Could not write metadata file: {error}") + self.fs.mkdir(dataset_path) + return ds + + def delete(self, dataset: Dataset) -> None: + dataset_path = self._get_path(dataset) + self.fs.rm(dataset_path, recursive=True) + metadata_file = dataset_path + ".json" + if self.fs.exists(metadata_file): + self.fs.rm(metadata_file) + + def delete_by_id(self, id: str) -> None: + raise ValueError("delete_by_id is ambiguous for FileBasedRepository. Provide a full dataset object.") + + def change_id(self, dataset: Dataset, new_id: str): + if dataset.metadata["id"] == new_id: + return + old_path = self._get_path(dataset) + new_path = old_path.replace(dataset.metadata["id"], new_id) + if self.fs.exists(new_path): + raise ValueError(f"New dataset ID {new_id} collides with an existing dataset path.") + self.fs.mv(old_path, new_path, recursive=True) + old_meta = old_path + ".json" + new_meta = new_path + ".json" + if self.fs.exists(old_meta): + self.fs.mv(old_meta, new_meta) + dataset.metadata["id"] = new_id + dataset.path = new_path + self.update_metadata(dataset) + + @measure_performance("upload", lambda self, dataset, local_path, local_file, filesystem=None: local_path) + def put(self, dataset: Dataset, local_path: str, rel_filename: str, filesystem: fsspec.AbstractFileSystem = None) -> str: + dataset_path = self._get_path(dataset) + if not self.fs.exists(dataset_path): + raise ValueError(f"Dataset {dataset.metadata['id']} does not exist at {dataset_path}") + remote_file = os.path.join(dataset_path, rel_filename) + remote_dir = os.path.dirname(remote_file) + if not self.fs.exists(remote_dir): + self.fs.mkdirs(remote_dir) + self.fs.put(local_path, remote_file) + return remote_file + + def copy_large_file(self, source_path: str, destination_path: str, source_fs, destination_fs): + try: + with source_fs.open(source_path, 'rb') as source_file: + with destination_fs.open(destination_path, 'wb') as destination_file: + while True: + chunk = source_file.read(self._chunk_size) + if not chunk: + break + destination_file.write(chunk) + except Exception as e: + logging.error(f"Failed to copy file: {e}") + raise + + async def copy_large_file_async(self, source_path: str, destination_path: str, source_fs, destination_fs): + try: + async with await asyncio.to_thread(source_fs.open, source_path, 'rb') as source_file: + async with await asyncio.to_thread(destination_fs.open, destination_path, 'wb') as destination_file: + while True: + chunk = await asyncio.to_thread(source_file.read, self._chunk_size) + if not chunk: + break + await asyncio.to_thread(destination_file.write, chunk) + except Exception as e: + logging.error(f"Failed to copy file: {e}") + raise + + @measure_performance("download", lambda self, dataset, file, local_path, filesystem=None: os.path.join(local_path, os.path.basename(file))) + def get(self, dataset: Dataset, file: str, local_path: str, filesystem: fsspec.AbstractFileSystem = None) -> None: + dataset_path = self._get_path(dataset) + source = file if file.startswith(dataset_path) else os.path.join(dataset_path, file) + if filesystem is None: + local_destination = os.path.join(local_path, os.path.basename(file)) + self.fs.get(source, local_destination) + else: + _rcnt = self._retry_count + while _rcnt >= 0: + try: + self.copy_large_file(source, os.path.join(local_path, os.path.basename(file)), self.fs, filesystem) + return + except Exception as e: + _rcnt -= 1 + if _rcnt <= 0: + raise e + + async def get_async(self, dataset: Dataset, file: str, local_path: str, filesystem: fsspec.AbstractFileSystem = None) -> None: + dataset_path = self._get_path(dataset) + source = file if file.startswith(dataset_path) else os.path.join(dataset_path, file) + if filesystem is None: + local_destination = os.path.join(local_path, os.path.basename(file)) + self.fs.get(source, local_destination) + else: + _rcnt = self._retry_count + while _rcnt >= 0: + try: + await self.copy_large_file_async(source, os.path.join(local_path, os.path.basename(file)), self.fs, filesystem) + return + except Exception as e: + _rcnt -= 1 + if _rcnt <= 0: + raise e + + def rm(self, dataset: Dataset, files: List[str]) -> None: + dataset_path = self._get_path(dataset) + for f in files: + self.fs.rm(os.path.join(dataset_path, f)) + + def checks( + self, + access: str = "public", + day: Union[str, datetime, None] = None, + duration: int = 0, + query=None, + cb_progress=None + ) -> list: + datasets = self.list(access, day, duration, query, cb_progress) + errors = [] + for dset in datasets: + path = self._get_path(dset) + if not self.fs.exists(path): + errors.append(f"Dataset {dset.metadata['id']} directory is missing at {path}") + meta_file = path + ".json" + if not self.fs.exists(meta_file): + errors.append(f"Metadata file missing for dataset {dset.metadata['id']}") + return errors + + def load_metadata(self, path: str) -> dict: + with self.fs.open(path, "r") as fh: + data = json.load(fh) + base_name = os.path.basename(path).replace(".json", "") + if "title" not in data: + data["title"] = "UNKNOWN TITLE" + if "id" not in data and "internalID" not in data and "dataset_id" not in data: + data["internalID"] = base_name + return data + + def update_metadata(self, dataset: Dataset, force: bool = False) -> list: + path = self._get_path(dataset) + if os.path.basename(path) != dataset.metadata["id"]: + return f"Dataset ID mismatch: {dataset.metadata['id']} vs {os.path.basename(path)}" + metadata_file = path + ".json" + try: + parent_dir = os.path.dirname(path) + if not self.fs.exists(parent_dir): + self.fs.mkdirs(parent_dir) + with self.fs.open(metadata_file, "w") as fh: + json.dump(dataset.metadata.as_json_dict(), fh) + except Exception as e: + if self.fs.exists(metadata_file): + self.fs.rm(metadata_file) + return str(e) + return None + + def readlines(self, dataset: Dataset, file_name: str) -> Union[str, None]: + path = self._get_path(dataset) + file_path = os.path.join(path, file_name) + if self.fs.exists(file_path): + with self.fs.open(file_path, "r") as file: + return file.read() + return None + + def writelines(self, dataset: Dataset, file_name: str, content: str) -> None: + path = self._get_path(dataset) + file_path = os.path.join(path, file_name) + dir_path = os.path.dirname(file_path) + if not self.fs.exists(dir_path): + self.fs.mkdirs(dir_path) + with self.fs.open(file_path, "w") as file: + file.write(content) + + +class S3FileBasedRepository(FileBasedRepository): + """S3-based repository.""" + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._backend_name = "s3-filesystem" + + +class LocalRepository(FileBasedRepository): + """A local filesystem repository.""" + pass diff --git a/owilix/core/repository/lexis.py b/owilix/core/repository/lexis.py new file mode 100644 index 0000000..d27e01b --- /dev/null +++ b/owilix/core/repository/lexis.py @@ -0,0 +1,338 @@ +""" +LEXIS Repository Implementation + +Provides LexisRepository for py4lexis 5.x using HTTP APIs only. + +Uses: +- OWILexisDatasetAPI for DDI metadata operations +- Http2IrodsFileSystem for file operations via iRODS HTTP API +""" + +import logging +import os +import time +from datetime import datetime +from typing import List, Union, Sequence + +import fsspec + +from owilix.core.repository.base import AbstractRepository +from owilix.core.metadata import Dataset + +logger = logging.getLogger("owilix") + + +class LexisRepository(AbstractRepository): + """ + No host/port/zone configuration needed - everything comes from LexisSession. + """ + + def __init__( + self, + manager, + lexis_project_id: str = None, + repo_name: str = "lexis", + tags: List[str] = None, + description: str = "LEXIS HTTP Repository (py4lexis 5.x)", + **kwargs + ): + """ + Initialize LexisRepository. + + Args: + manager: OWILIXManager instance with session + lexis_project_id: Override project ID (default: from env) + repo_name: Repository name for identification + tags: Tags for filtering + description: Human-readable description + """ + super().__init__() + self.manager = manager + self._lexis_project_id = lexis_project_id + self._repo_name = repo_name + self._tags = tags if tags is not None else ["lexis", "remote"] + self._description = description + self._backend_name = "lexis-http" + + # Lazy-initialized components + self._ddi_api = None + self._filesystem = None + self._zone = kwargs.get("zone", None) # Optional zone override + + @property + def lexis_session(self): + """Get LexisSession from manager.""" + return self.manager.session.lexis + + @property + def ddi_api(self): + """Lazy-init DDI API wrapper.""" + if self._ddi_api is None: + from owilix.core.lexis import OWILexisDatasetAPI + self._ddi_api = OWILexisDatasetAPI( + self.lexis_session, + suppress_print=True + ) + return self._ddi_api + + @property + def fs(self): + """Lazy-init Http2IrodsFileSystem.""" + if self._filesystem is None: + from owilix.core.fsspec import Http2IrodsFileSystem + from py4lexis.core.lexis_irods import iRODS + + # Create iRODS wrapper to get access to the client + # The LexisSession handles authentication + wrapper = iRODS(self.lexis_session) + + # Access the underlying irods_http_client instance + # Based on py4lexis internals (referenced in tests) + client = getattr(wrapper, 'irods_client', getattr(wrapper, '_irds', None)) + + if client is None: + # Fallback: check mangled name + client = getattr(wrapper, '_iRODS__irds', None) + + if client is None: + raise RuntimeError("Could not retrieve iRODS HTTP client from py4lexis wrapper") + + # Get base URL from client or session + irods_url = getattr(client, 'url_base', self.lexis_session.irods_http_api_url) + + self._filesystem = Http2IrodsFileSystem( + irods_client=client, + url_base=irods_url + ) + return self._filesystem + + @property + def zone(self) -> str: + """Get iRODS zone from session or override.""" + if self._zone: + return self._zone + # Extract zone from session URL or use default + # The zone is typically part of the iRODS path structure + return getattr(self.lexis_session, '_zone', 'IT4ILexisV2') + + def _get_irods_path(self, access: str, dataset_id: str = None) -> str: + """Build iRODS path for a dataset.""" + base = f"/{self.zone}/{access}" + if dataset_id: + return f"{base}/{dataset_id}" + return base + + def _get_path(self, dataset: Dataset) -> str: + """Get iRODS path for a dataset.""" + if dataset.path: + return dataset.path + return self._get_irods_path(dataset.access, dataset.metadata["id"]) + + def status(self) -> dict: + """Check repository connectivity.""" + try: + # Test DDI API connectivity + self.ddi_api.get_all_datasets(content_as_pandas=False) + return self._add_details_to_status({ + "status": True, + "message": "connected", + "public": True, "user": True, "project": True + }) + except Exception as e: + return self._add_details_to_status({ + "status": False, + "message": f"Connection failed: {str(e)}", + "public": False, "user": False, "project": False + }) + + def exists(self, dataset: Dataset) -> bool: + """Check if dataset exists.""" + try: + path = self._get_path(dataset) + return self.fs.exists(path) + except Exception: + return False + + def list( + self, + access: str = "public", + day: Union[str, datetime, None] = None, + duration: int = 0, + query=None, + cb_progress=None + ) -> List[Dataset]: + """List datasets from LEXIS DDI API.""" + start_time = time.time() + + try: + # Use DDI API to get datasets - pass project name from manager + result = self.ddi_api.get_all_datasets( + access=access, + project=getattr(self.manager, 'name', None), + content_as_pandas=False + ) + + datasets = [] + items = result if isinstance(result, list) else [] + + for record in items: + try: + # Pass the full record to Dataset like old implementation + # This preserves all metadata fields (startDate, endDate, objectCount, etc.) + ds = Dataset( + repository=self, + path=record.get("absolute_path", self._get_irods_path(access, record.get("dataset_id", record.get("id", "")))), + **record + ) + datasets.append(ds) + except Exception as e: + logger.warning(f"Error building dataset from record: {e}") + + # Update performance stats + elapsed = time.time() - start_time + self.update_performance_stats("list", elapsed, len(datasets)) + + return Dataset.filter_datasets(datasets, day, duration, query) + + except Exception as e: + logger.error(f"Failed to list datasets: {e}") + return [] + + def files(self, dataset: Dataset, files_glob: str | Sequence[str] = None) -> list: + """List files in a dataset.""" + path = self._get_path(dataset) + + if files_glob is None: + patterns = ["**/*"] + elif isinstance(files_glob, str): + patterns = [files_glob] + else: + patterns = list(files_glob) + + hits = set() + for pat in patterns: + full_pattern = os.path.join(path, pat) + try: + result = self.fs.glob(full_pattern) + for p in result: + if not self.fs.isdir(p): + hits.add(p) + except Exception as e: + logger.warning(f"Glob pattern {full_pattern} failed: {e}") + + return sorted(hits) + + def files_details( + self, + dataset: Dataset, + files_glob: str | Sequence[str] = None, + count_rows: bool = False + ) -> list: + """Get file details with optional row counting.""" + from owilix.core.utils import fill_file_details + file_list = self.files(dataset, files_glob) + return fill_file_details(self.fs, file_list, root=self._get_path(dataset), do_count=count_rows) + + def readlines(self, dataset: Dataset, file_name: str) -> Union[str, None]: + """Read file content as text.""" + path = os.path.join(self._get_path(dataset), file_name) + try: + with self.fs.open(path, "r") as f: + return f.read() + except Exception: + return None + + def writelines(self, dataset: Dataset, file_name: str, content: str) -> None: + """Write text content to file.""" + path = os.path.join(self._get_path(dataset), file_name) + with self.fs.open(path, "w") as f: + f.write(content) + + def delete(self, dataset: Dataset) -> None: + """Delete a dataset via DDI API.""" + # Note: Deletion typically goes through DDI API, not direct iRODS + raise NotImplementedError("Dataset deletion should use DDI API") + + def delete_by_id(self, id: str) -> None: + """Delete dataset by ID.""" + raise NotImplementedError("Dataset deletion should use DDI API") + + def put( + self, + dataset: Dataset, + local_path: str, + rel_filename: str, + filesystem: fsspec.AbstractFileSystem = None + ) -> str: + """Upload a file to dataset.""" + remote_path = os.path.join(self._get_path(dataset), rel_filename) + remote_dir = os.path.dirname(remote_path) + + # Ensure directory exists + if not self.fs.exists(remote_dir): + self.fs.makedirs(remote_dir) + + # Upload file + start_time = time.time() + file_size = os.path.getsize(local_path) + + self.fs.put(local_path, remote_path) + + elapsed = time.time() - start_time + self.update_performance_stats("upload", elapsed, file_size) + + return remote_path + + def get( + self, + dataset: Dataset, + file: str, + local_path: str, + filesystem: fsspec.AbstractFileSystem = None + ) -> None: + """Download a file from dataset.""" + remote_path = file if file.startswith("/") else os.path.join(self._get_path(dataset), file) + local_file = os.path.join(local_path, os.path.basename(file)) + + start_time = time.time() + + self.fs.get(remote_path, local_file) + + elapsed = time.time() - start_time + file_size = os.path.getsize(local_file) if os.path.exists(local_file) else 0 + self.update_performance_stats("download", elapsed, file_size) + + def rm(self, dataset: Dataset, files: List[str]) -> None: + """Remove files from dataset.""" + base_path = self._get_path(dataset) + for f in files: + path = f if f.startswith("/") else os.path.join(base_path, f) + self.fs.rm(path) + + def create( + self, + access: str = "public", + collectionName: str = "main", + metadata=None + ) -> Dataset: + """Create a new dataset via DDI API.""" + # Dataset creation should go through py4lexis DDI API + raise NotImplementedError("Dataset creation should use DDI API directly") + + def change_id(self, dataset: Dataset, new_id: str): + """Change dataset ID (not supported via HTTP).""" + raise NotImplementedError("ID changes not supported for LEXIS datasets") + + def update_metadata(self, dataset: Dataset, force: bool = False) -> list: + """Update dataset metadata via DDI API.""" + try: + self.ddi_api.update_dataset_metadata( + dataset_id=dataset.metadata["id"], + metadata=dataset.metadata.as_json_dict() + ) + return [] + except Exception as e: + return [str(e)] + + def __str__(self): + return f"LexisRepository(zone={self.zone})" diff --git a/tests/owilix/core/fsspec/test_core_fsspec_integration.py b/tests/owilix/core/fsspec/test_core_fsspec_integration.py index ccca9aa..1d4fd68 100644 --- a/tests/owilix/core/fsspec/test_core_fsspec_integration.py +++ b/tests/owilix/core/fsspec/test_core_fsspec_integration.py @@ -13,89 +13,38 @@ import pytest # Skip all tests in this module if no token available pytestmark = pytest.mark.integration - -def get_session(): - """ - Get LexisSession using refresh token pattern from small_tst_py4lexis.py. - - Stores token in ~/tmp/refresh_token.txt for reuse. - """ - from py4lexis.session import LexisSession - - token_path = os.path.expanduser("~/tmp/refresh_token.txt") - refresh_token = None - - try: - with open(token_path, "r") as f: - refresh_token = f.read().strip() - except FileNotFoundError: - pass - - if refresh_token: - session = LexisSession(login_method="token", refresh_token=refresh_token) - else: - session = LexisSession() - - # Store token for next run - os.makedirs(os.path.dirname(token_path), exist_ok=True) - with open(token_path, "w") as f: - f.write(session.get_refresh_token()) - - return session - - -def get_irods_client(session): - """ - Get iRODS HTTP client with token refresh wrapper. - - Based on OWIIrods pattern from small_tst_py4lexis.py. - """ - from py4lexis.core.lexis_irods import iRODS - - class OWIIrods(iRODS): - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - - def irods(self): - """Get iRODS client with token refresh check.""" - self._iRODS__check_access_token() - return self._irds - - return OWIIrods(session=session, suppress_print=True) - - -# Test dataset from small_tst_py4lexis.py +# Test dataset used for integration tests TEST_DATASET_ID = "f6ea5756-2e0b-11ef-b336-0242ac1d0004" @pytest.fixture(scope="module") -def session(): - """Create session once per module.""" - return get_session() - +def lexis_repo(): + """Get LexisRepository using OWIlixManager.""" + from owilix.core.manager import OWIlixManager + from owilix.core.repository.lexis import LexisRepository + from owilix.core.lexis import OWILexisDatasetAPI + + manager = OWIlixManager() + return LexisRepository(manager) @pytest.fixture(scope="module") -def irods_client(session): - """Create iRODS client once per module.""" - return get_irods_client(session) - +def fs(lexis_repo): + """Get Http2IrodsFileSystem from repository.""" + return lexis_repo.fs @pytest.fixture(scope="module") -def fs(irods_client): - """Create Http2IrodsFileSystem once per module.""" - from owilix.core.fsspec.http2irods import Http2IrodsFileSystem +def test_collection(lexis_repo): + """Get test collection path.""" + # We need the underlying irods wrapper to get collection + # LexisRepository doesn't expose get_dataset_collection directly - return Http2IrodsFileSystem( - irods_client=irods_client.irods(), - url_base=irods_client.irods().url_base, - ) + from py4lexis.core.lexis_irods import iRODS + wrapper = iRODS(lexis_repo.lexis_session) + coll = wrapper.get_dataset_collection(TEST_DATASET_ID) + return coll + -@pytest.fixture(scope="module") -def test_collection(irods_client): - """Get test collection path.""" - coll = irods_client.get_dataset_collection(TEST_DATASET_ID) - return coll class TestHttp2IrodsIntegration: diff --git a/tests/owilix/core/repository/__init__.py b/tests/owilix/core/repository/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/owilix/core/repository/test_integration.py b/tests/owilix/core/repository/test_integration.py new file mode 100644 index 0000000..7f275ed --- /dev/null +++ b/tests/owilix/core/repository/test_integration.py @@ -0,0 +1,108 @@ +import pytest +from click.testing import CliRunner +import owilix.cli +import os +import yaml + +class TestIntegration: + @pytest.fixture(autouse=True) + def setup_cli(self): + # Register commands for the CLI runner + owilix.cli.register_commands(owilix.cli.cli) + + @pytest.fixture + def runner(self): + return CliRunner() + + def test_remote_ls(self, runner): + """Test listing remote datasets.""" + result = runner.invoke(owilix.cli.cli, ['--loglevel', 'DEBUG', 'remote', 'ls', 'all'], prog_name='owi') + assert result.exit_code == 0 + assert "Fetching datasets" in result.output + + def test_remote_pull_specific_dataset(self, runner, tmp_path): + """Test pulling a specific dataset. + Command: owi --yes remote pull all/id=0350fecc-e58b-11f0-a8c9-8ebf6bb2cab9 + """ + target_dir = str(tmp_path) + + # Setup configuration in target_dir to reuse credentials + # This prevents login prompt by copying existing token from ~/.owi/owilix.cfg + from owilix.core.manager import OWIlixManager + import yaml + + # Load default config (from ~/.owi) + mgr = OWIlixManager() + cfg = mgr.config.config.copy() + + # Ensure repositories structure exists and override local path + if "repositories" not in cfg: cfg["repositories"] = {} + if "config" not in cfg["repositories"]: cfg["repositories"]["config"] = {} + if "local" not in cfg["repositories"]["config"]: + cfg["repositories"]["config"]["local"] = {"repository":"file", "options":{}} + + # Override path to use the temp directory + # The 'target' in OWIlixManager is base_path, but config might have other paths. + # We explicitly set it to tmp_path/{access} + cfg["repositories"]["config"]["local"]["options"]["path"] = os.path.join(target_dir, "{access}") + + # Save to target_dir/owilix.cfg + with open(os.path.join(target_dir, "owilix.cfg"), "w") as f: + yaml.dump(cfg, f) + + # Copy refresh token if available + src_token_path = mgr._refresh_token_fn + if os.path.exists(src_token_path): + dst_token_dir = os.path.join(target_dir, ".tokens") + if not os.path.exists(dst_token_dir): + os.makedirs(dst_token_dir) + import shutil + shutil.copy2(src_token_path, os.path.join(dst_token_dir, "refresh_token")) + + # Use --target to pull to a temp directory + + + + # Use --target to pull to a temp directory + # Using ID from test_core_fsspec_integration which is known to work + # ID: f6ea5756-2e0b-11ef-b336-0242ac1d0004 + args = [ + '--loglevel', 'DEBUG', + '--yes', + '--target', target_dir, + 'remote', 'pull', + 'all/id=f6ea5756-2e0b-11ef-b336-0242ac1d0004', + 'overwrite=true' + ] + + result = runner.invoke(owilix.cli.cli, args, prog_name='owi', input="yes\n") + + # If this fails (like currently), print output for debugging + if result.exit_code != 0: + print(f"Command Output:\n{result.output}") + + assert result.exit_code == 0 + assert "Fetching files" in result.output + + # Verify local availability using local ls + # We must use the same target directory so it finds the repository + # local ls uses 'all' specifier to list everything + result_ls = runner.invoke(owilix.cli.cli, + ['--target', target_dir, '--loglevel', 'DEBUG', 'local', 'ls', 'all', 'files=**/*'], + prog_name='owi') + + if result_ls.exit_code != 0: + print(f"Local LS Output:\n{result_ls.output}") + + assert result_ls.exit_code == 0 + # Check if it found the dataset and files + assert "Fetching datasets" in result_ls.output + assert "Found 22" in result.output or "Files downloaded" in result.output # From pull + # For ls output processing: + # It should print file details if files=**/* is used + # Check for some known file or just count + assert "index.ciff.gz" in result_ls.output # Based on previous output + + + # We can also check if files were downloaded if we knew filenames + # But just checking success code is good initial integration test diff --git a/tests/owilix/core/repository/test_repository.py b/tests/owilix/core/repository/test_repository.py new file mode 100644 index 0000000..ed3a17c --- /dev/null +++ b/tests/owilix/core/repository/test_repository.py @@ -0,0 +1,119 @@ +""" +Tests for owilix.core.repository package + +Tests for repository classes and imports. +""" + +import pytest + + +class TestRepositoryImports: + """Test repository package imports.""" + + def test_import_base_classes(self): + """Test base class imports.""" + from owilix.core.repository import ( + AbstractRepository, + PerformanceMetric, + RepoPerformanceStats, + ) + assert AbstractRepository is not None + assert PerformanceMetric is not None + assert RepoPerformanceStats is not None + + def test_import_file_repository(self): + """Test FileBasedRepository import.""" + from owilix.core.repository import FileBasedRepository, LocalRepository + assert FileBasedRepository is not None + assert LocalRepository is not None + + def test_import_lexis_repository(self): + """Test LexisRepository import.""" + from owilix.core.repository import LexisRepository + assert LexisRepository is not None + + def test_import_aggregated_repository(self): + """Test AggregatedRepository import.""" + from owilix.core.repository import AggregatedRepository + assert AggregatedRepository is not None + + def test_legacy_alias_lexis_irods(self): + """Test legacy LEXISIrodsRepository alias.""" + from owilix.core.repository import LEXISIrodsRepository, LexisRepository + assert LEXISIrodsRepository is LexisRepository + + +class TestPerformanceMetric: + """Tests for PerformanceMetric class.""" + + def test_metric_initialization(self): + """Test PerformanceMetric initialization.""" + from owilix.core.repository import PerformanceMetric + metric = PerformanceMetric() + assert metric.ema is None + assert metric.buffer == [] + assert metric.count == 0 + + def test_metric_update(self): + """Test PerformanceMetric update.""" + from owilix.core.repository import PerformanceMetric + metric = PerformanceMetric() + metric.update(1.0) + metric.update(2.0) + assert metric.count == 2 + assert len(metric.buffer) == 2 + + def test_metric_serialization(self): + """Test PerformanceMetric to/from JSON.""" + from owilix.core.repository import PerformanceMetric + metric = PerformanceMetric() + metric.update(1.0) + + data = metric.to_json() + restored = PerformanceMetric.from_json(data) + + assert restored.count == metric.count + assert restored.buffer == metric.buffer + + +class TestRepoPerformanceStats: + """Tests for RepoPerformanceStats class.""" + + def test_stats_initialization(self): + """Test RepoPerformanceStats initialization.""" + from owilix.core.repository import RepoPerformanceStats + stats = RepoPerformanceStats() + assert stats.list_response_time is not None + assert stats.list_bandwidth is not None + assert stats.upload_bandwidth is not None + assert stats.download_bandwidth is not None + + def test_stats_serialization(self): + """Test RepoPerformanceStats to/from JSON.""" + from owilix.core.repository import RepoPerformanceStats + stats = RepoPerformanceStats() + stats.list_response_time.update(0.5) + + data = stats.to_json() + restored = RepoPerformanceStats.from_json(data) + + assert restored.list_response_time.count == stats.list_response_time.count + + +class TestCoreExports: + """Test core package exports repository classes.""" + + def test_core_exports_local_repository(self): + """Test LocalRepository accessible from core.""" + from owilix.core import LocalRepository + assert LocalRepository is not None + + def test_core_exports_aggregated_repository(self): + """Test AggregatedRepository accessible from core.""" + from owilix.core import AggregatedRepository + assert AggregatedRepository is not None + + def test_core_exports_lexis_repository(self): + """Test LexisRepository accessible from core.""" + from owilix.core import LexisRepository + assert LexisRepository is not None diff --git a/tests/owilix/core/repository/test_repository_legacy_ported.py b/tests/owilix/core/repository/test_repository_legacy_ported.py new file mode 100644 index 0000000..4c68500 --- /dev/null +++ b/tests/owilix/core/repository/test_repository_legacy_ported.py @@ -0,0 +1,211 @@ + +# test_repository_legacy_ported.py +# Migrated from tests/owilix/test_repositories.py + +import json +import os +import time +from dataclasses import dataclass +from typing import Dict, List, Sequence, Union, Any, Optional + +import pytest + +# Use new repository package imports +from owilix.core.repository import ( + PerformanceMetric, + RepoPerformanceStats, + FileBasedRepository, + AggregatedRepository, + AbstractRepository +) + +# --- Minimal dataset stub the module methods can work with --- + +class DummyMetaDict(dict): + """Expose as_json_dict() like the real Dataset.metadata might do in update_metadata().""" + def as_json_dict(self): + return dict(self) + + +@dataclass +class DummyDataset: + repository: Any + path: Optional[str] + id: str + access: str = "public" + collectionName: str = "main" + + @property + def metadata(self) -> DummyMetaDict: + return DummyMetaDict({"id": self.id}) + + @classmethod + def filter_datasets(cls, datasets, day=None, duration=0, query=None): + return datasets + + +# ---------- PerformanceMetric tests ---------- +def test_performance_metric_update_and_overall_average(): + m = PerformanceMetric(alpha=0.5, buffer_size=3, weight_ema=0.6) + + values = [10, 20, 30, 40] + for v in values: + m.update(v) + + # EMA progression 0.5: 10 -> 15 -> 22.5 -> 31.25 + assert pytest.approx(m.ema, rel=1e-6) == 31.25 + + # buffer=[20,30,40] avg=30 + assert m.buffer == [20, 30, 40] + assert pytest.approx(m.get_buffer_average(), rel=1e-6) == 30.0 + + # overall = 0.6*31.25 + 0.4*30 = 30.75 + assert pytest.approx(m.get_overall_average(), rel=1e-6) == 30.75 + + # json round trip + data = m.to_json() + m2 = PerformanceMetric.from_json(data) + assert m2.alpha == m.alpha + assert m2.buffer_size == m.buffer_size + + +def test_repo_performance_stats_roundtrip(): + stats = RepoPerformanceStats() + stats.list_response_time.update(1.0) + stats.list_bandwidth.update(100.0) + + j = stats.to_json() + stats2 = RepoPerformanceStats.from_json(j) + assert pytest.approx(stats2.list_response_time.get_overall_average(), rel=1e-6) == \ + stats.list_response_time.get_overall_average() + + +# ---------- FileBasedRepository: files() behavior ---------- +def make_local_repo(tmp_path) -> FileBasedRepository: + base = os.path.join(str(tmp_path), "{access}") + return FileBasedRepository( + manager=None, + path=base, + repo_name="local", + protocol="file", + collections_in_path=True, + ) + +def make_dataset_for_repo(repo: FileBasedRepository, ds_id="abc123", access="public", collection="main"): + # Ensure the dataset directory exists under repo path + dataset_dir = os.path.join(repo.path.format(access=access), collection, ds_id) + os.makedirs(dataset_dir, exist_ok=True) + return DummyDataset(repository=repo, path=dataset_dir, id=ds_id, access=access, collectionName=collection) + +def test_filebased_files_globs_and_absolute(tmp_path): + repo = make_local_repo(tmp_path) + ds = make_dataset_for_repo(repo) + + # create files + root = ds.path + f1 = os.path.join(root, "a.txt") + os.makedirs(os.path.join(root, "sub"), exist_ok=True) + f2 = os.path.join(root, "sub", "b.parquet") + f3 = os.path.join(root, "sub", "c.bin") + for p in (f1, f2, f3): + with open(p, "wb") as fh: + fh.write(b"x") + + # default: all files + files = repo.files(ds, None) + assert set(files) == {f1, f2, f3} + + # single glob + only_txt = repo.files(ds, "*.txt") + assert only_txt == [f1] + + # absolute-like pattern + abs_match = repo.files(ds, f3) + assert abs_match == [f3] + + +# ---------- measure_performance decorator (on put/get) ---------- +def test_put_and_get_update_bandwidth(tmp_path): + repo = make_local_repo(tmp_path) + ds = make_dataset_for_repo(repo) + + src_dir = tmp_path / "src" + dl_dir = tmp_path / "dl" + src_dir.mkdir(parents=True, exist_ok=True) + dl_dir.mkdir(parents=True, exist_ok=True) + + local_src = src_dir / "file.bin" + with open(local_src, "wb") as fh: + fh.write(b"x" * 1024) + + # upload => check stats + remote_path = repo.put(ds, str(local_src), "data/file.bin") + assert repo.fs.exists(remote_path) + assert repo.performance_stats.upload_bandwidth.count >= 1 + + # download => check stats + repo.get(ds, "data/file.bin", str(dl_dir)) + local_dl = dl_dir / "file.bin" + assert local_dl.exists() + assert repo.performance_stats.download_bandwidth.count >= 1 + + +# ---------- AggregatedRepository: dedup by performance & stats store/load ---------- +class TinyRepo(AbstractRepository): + """Minimal concrete repo.""" + def __init__(self, name: str, score_bias: float = 0.0): + super().__init__() + self._repo_name = name + self._backend_name = "tiny" + self.ds_id = "same-id" + self.bias = score_bias + + def list(self, access="public", day=None, duration=0, query=None, cb_progress=None) -> List[DummyDataset]: + ds = DummyDataset(repository=self, path=None, id=self.ds_id, access=access, collectionName="main") + time.sleep(0.01) + return [ds] + + # Implement abstract methods + def exists(self, dataset): return True + def files(self, dataset, files_glob=None): return [] + def files_details(self, dataset, files_glob=None, count_rows=False): return [] + def readlines(self, dataset, file_name): return "" + def writelines(self, dataset, file_name, content): pass + def delete(self, dataset): pass + def delete_by_id(self, id): pass + def put(self, dataset, local_path, rel_filename, filesystem=None): return "" + def get(self, dataset, file, local_path, filesystem=None): pass + def rm(self, dataset, files): pass + def create(self, access="public", collectionName="main", metadata=None): raise NotImplementedError + def change_id(self, dataset, new_id): pass + + +def test_aggregated_repo_picks_best_by_stats(tmp_path): + r_fast = TinyRepo("fast") + r_slow = TinyRepo("slow") + + # Make "fast" better + r_fast.performance_stats.list_response_time.update(0.05) + r_fast.performance_stats.list_bandwidth.update(100.0) + + # Make "slow" worse + r_slow.performance_stats.list_response_time.update(1.0) + r_slow.performance_stats.list_bandwidth.update(1.0) + + agg = AggregatedRepository({"fast": r_fast, "slow": r_slow}) + + best = agg.list() + assert len(best) == 1 + assert best[0].repository is r_fast + + # Store stats + stats_file = os.path.join(tmp_path, "stats.json") + agg.store_stats_to_file(stats_file) + assert os.path.exists(stats_file) + + # Reload + r_fast.performance_stats = RepoPerformanceStats() + agg.load_stats_from_file(stats_file) + loaded = agg.get_performance_stats() + assert "fast" in loaded + assert loaded["fast"]["list_bandwidth"]["count"] >= 1