GH45 - add xts alias rework #52
Conversation
|
I have read the CLA Document and I hereby sign the CLA 1 out of 2 committers have signed the CLA. |
There was a problem hiding this comment.
Pull request overview
This PR significantly refactors the XTS alias management system to support multiple input types (URLs, files, and directories) with improved caching and a cleaner CLI interface.
Key changes:
- Rewrote alias management to handle URLs, single files, and directories of .xts files
- Changed CLI from
xts aliassubcommands toxts --aliasflag-based interface - Simplified the main XTS class by removing auto-discovery of .xts files in favor of explicit alias resolution
- Introduced hash-based cache naming to avoid file collisions
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 10 comments.
| File | Description |
|---|---|
| src/xts_core/xts_alias.py | Complete rewrite with new helper functions, improved caching logic, directory support, and CLI handler |
| src/xts_core/xts.py | Removed old alias subcommand parsing, removed auto-discovery, simplified to use alias-only workflow with new help system |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| ensure_dirs() | ||
| filename = hashlib.sha256(url.encode()).hexdigest() + ".xts" | ||
| path = os.path.join(CACHE_DIR, filename) | ||
|
|
||
| if not os.path.exists(path): | ||
| print(f"Fetching remote .xts config from: {url}") | ||
| r = requests.get(url) | ||
| r.raise_for_status() | ||
| with open(path, "w", encoding="utf-8") as f: | ||
| f.write(r.text) | ||
| # filename = hashlib.sha256(url.encode()).hexdigest() + ".xts" | ||
| filename = Path(urlparse(url).path).name | ||
| if not filename: | ||
| raise ValueError(f"URL does not contain a filename: {url}") | ||
|
|
||
| cached_path = os.path.join(CACHE_DIR, filename) | ||
|
|
||
| print(f"Fetching remote .xts config from: {url}") | ||
| r = requests.get(url) | ||
| r.raise_for_status() | ||
|
|
||
| with open(cached_path, "w", encoding="utf-8") as f: | ||
| f.write(r.text) | ||
|
|
||
| return path | ||
| return cached_path |
There was a problem hiding this comment.
The fetch_url_to_cache function now always re-downloads the file even if it already exists in cache (the if not os.path.exists(path) check was removed). This could lead to unnecessary network requests and slower performance. Consider adding back a check to skip downloading if the file already exists, or add a force parameter to control this behavior.
src/xts_core/xts_alias.py
Outdated
|
|
||
| def _cache_name_for_local_file(src: Path) -> str: | ||
| """ | ||
| Produce a stable cache filename for a local file, avoiding collisoins |
There was a problem hiding this comment.
Spelling error: "collisoins" should be "collisions".
| Produce a stable cache filename for a local file, avoiding collisoins | |
| Produce a stable cache filename for a local file, avoiding collisions |
src/xts_core/xts_alias.py
Outdated
| cache_name = _cache_name_for_local_file(src) | ||
| dst = Path(CACHE_DIR) / cache_name | ||
|
|
||
| shutil.copy2(src, dst) #ovewrite to refresh |
There was a problem hiding this comment.
Spelling error: "#ovewrite" should be "# overwrite" (missing space after #).
| shutil.copy2(src, dst) #ovewrite to refresh | |
| shutil.copy2(src, dst) # overwrite to refresh |
src/xts_core/xts_alias.py
Outdated
| if remove_name is not None: | ||
| removed = remove_alias(remove_name) | ||
| if removed: | ||
| print(f"Removed alias: {remove_name}") | ||
| return 0 | ||
| print(f"Alias not found: {remove_name}") | ||
| return 2 |
There was a problem hiding this comment.
The variable 'removed' is assigned the return value of remove_alias(), but remove_alias() returns None. This check will always fail. The logic should be revised to check whether the alias exists before attempting removal, or the remove_alias function should return a boolean.
src/xts_core/xts_alias.py
Outdated
| def remove_alias(name: str) -> None: | ||
| """ | ||
| Remove an alias if it exists. | ||
| """ | ||
| aliases = load_aliases() | ||
| if name in aliases: | ||
| del aliases[name] | ||
| with open(ALIAS_FILE, "w") as f: | ||
| json.dump(aliases, f, indent=2) | ||
| save_aliases(aliases) |
There was a problem hiding this comment.
The remove_alias function returns None in all cases, but the calling code checks if the return value is truthy (line 340). This will always evaluate to False, making the success message unreachable. The function should return True when an alias is removed or False when it doesn't exist.
src/xts_core/xts.py
Outdated
| if not resolved_xts_path: | ||
| error(f"Alias '{alias_name}' not found. Add it via: xts --alias <path|url|dir> [--name <name>]") | ||
|
|
||
| # laod xts config remove alias name from argv before parsing |
There was a problem hiding this comment.
Spelling error: "laod" should be "load".
| # laod xts config remove alias name from argv before parsing | |
| # load xts config remove alias name from argv before parsing |
src/xts_core/xts_alias.py
Outdated
| from xts_core import utils | ||
|
|
||
| try: | ||
| from .xts_arg_parser import XTSArgumentParser |
There was a problem hiding this comment.
Extra space between 'from' and '.xts_arg_parser'. Should be 'from .xts_arg_parser import XTSArgumentParser'.
| from .xts_arg_parser import XTSArgumentParser | |
| from .xts_arg_parser import XTSArgumentParser |
src/xts_core/xts_alias.py
Outdated
|
|
||
| def _cache_name_for_local_file(src: Path) -> str: | ||
| """ | ||
| Produce a stable cache filename for a local file, avoiding collisoins |
There was a problem hiding this comment.
Corrected spelling of 'collisoins' to 'collisions'.
| Produce a stable cache filename for a local file, avoiding collisoins | |
| Produce a stable cache filename for a local file, avoiding collisions |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 2 out of 2 changed files in this pull request and generated 20 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
src/xts_core/xts_alias.py
Outdated
| os.makedirs(CACHE_DIR, exist_ok=True) | ||
| os.makedirs(os.path.dirname(ALIAS_FILE), exist_ok=True) | ||
|
|
||
| # move these to utils |
There was a problem hiding this comment.
The comment on line 60 "move these to utils" suggests these helper functions should be relocated. If the intention is to move _is_url to the existing utils.is_url function, this creates duplication. The existing utils.is_url uses a simpler string startswith check, while _is_url uses urlparse with more robust validation. Consider consolidating these into a single implementation in utils.py or removing the comment if the functions should remain here.
| # move these to utils | |
| # Local helper functions; _is_url intentionally differs from utils.is_url | |
| # (uses urllib.parse.urlparse for stricter URL validation). |
src/xts_core/xts_alias.py
Outdated
| action='store', | ||
| default=None, | ||
| help='URI of xts file to add alias of', | ||
| nargs=argparse.OPTIONAL) |
There was a problem hiding this comment.
The nargs=argparse.OPTIONAL should be nargs='?' for an optional positional argument in argparse. argparse.OPTIONAL is not a valid constant; the correct value is the string '?'.
| nargs=argparse.OPTIONAL) | |
| nargs='?') |
src/xts_core/xts_alias.py
Outdated
| cache_name = _cache_name_for_local_file(src) | ||
| dst = Path(CACHE_DIR) / cache_name | ||
|
|
||
| shutil.copy2(src, dst) #overwrite to refresh |
There was a problem hiding this comment.
The inline comment shows lack of proper spacing around the '#' character. According to PEP 8, there should be at least two spaces before an inline comment and one space between the '#' and the comment text. The comment should be: '# overwrite to refresh'
| shutil.copy2(src, dst) #overwrite to refresh | |
| shutil.copy2(src, dst) # overwrite to refresh |
| return True | ||
| return False | ||
|
|
||
| def resolve_alias_to_xts_path(alias_name: str) -> str | None: |
There was a problem hiding this comment.
The return type annotation 'str | None' uses the Python 3.10+ union syntax. If the project needs to support Python 3.9 or earlier, this should use 'Optional[str]' from the typing module instead. Check the project's Python version requirements to ensure compatibility.
src/xts_core/xts.py
Outdated
| remaining_args.append(args.alias) | ||
| xts_alias.run_alias_builtin(remaining_args) | ||
| raise SystemExit(0) | ||
| resolved_xts_path = xts_alias.resolve_alias_to_xts_path(args.alias) |
There was a problem hiding this comment.
The resolved_xts_path may be None if the alias is not found in resolve_alias_to_xts_path. Setting self.xts_config to None will cause an error in the xts_config setter at line 130 ("xts config specified does not exist"). Consider adding explicit validation and providing a more helpful error message to the user about the unknown alias.
| resolved_xts_path = xts_alias.resolve_alias_to_xts_path(args.alias) | |
| resolved_xts_path = xts_alias.resolve_alias_to_xts_path(args.alias) | |
| if resolved_xts_path is None: | |
| # Provide a clear error when the alias cannot be resolved instead of | |
| # passing None into the xts_config setter. | |
| error( | |
| f'Unknown alias "{args.alias}". ' | |
| 'Use "xts --alias list" to see available aliases.' | |
| ) | |
| raise SystemExit(1) |
src/xts_core/xts_alias.py
Outdated
| cache_name = _cache_name_for_local_file(src) | ||
| dst = Path(CACHE_DIR) / cache_name | ||
|
|
||
| shutil.copy2(src, dst) #overwrite to refresh |
There was a problem hiding this comment.
The comment indicates 'overwrite to refresh', but shutil.copy2 preserves file metadata including modification times from the source. If the intention is to always refresh with the latest source content, this is correct. However, if modification times are important for cache invalidation logic, consider documenting this behavior more explicitly or using shutil.copy instead.
| shutil.copy2(src, dst) #overwrite to refresh | |
| shutil.copy(src, dst) # overwrite to refresh cache content and metadata |
| filename = Path(urlparse(url).path).name | ||
| if not filename: | ||
| raise ValueError(f"URL does not contain a filename: {url}") | ||
|
|
There was a problem hiding this comment.
The commented-out line suggests the previous implementation used a SHA-256 hash of the URL as the filename. The new implementation uses the actual filename from the URL path. This change creates a potential security issue: filenames from URLs could contain path traversal characters (e.g., "../evil.xts") or other problematic characters. Consider sanitizing the filename or validating that it doesn't contain directory separators before using it.
| # Validate that the derived filename cannot be used for path traversal. | |
| if filename in (".", ".."): | |
| raise ValueError(f"URL contains an invalid filename: {url}") | |
| for sep in (os.sep, os.path.altsep): | |
| if sep and sep in filename: | |
| raise ValueError(f"URL contains an invalid filename: {url}") |
| sys.exit(sorted(exit_code)[-1]) | ||
| except Exception as e: | ||
| error( | ||
| 'An unrecognised command caused an error\n\n' |
There was a problem hiding this comment.
The error message uses 'An unrecognised command' but the word is misspelled. It should be 'An unrecognized command' (American) or 'An unrecognised command' (British). The codebase appears to use American spelling based on other comments, so 'unrecognized' is preferred.
| def _parse_first_arg(self): | ||
| """ | ||
| Parse CLI arguments and set up argparse for all commands. | ||
| The first argument must be either: | ||
| - a built-in option starting with '--' (currently supported: --alias) | ||
| - an alias name (resolved via ~/.xts/aliases.json to an .xts file path) | ||
| Direct .xts file usage from cwd or as the first argument is not supported. |
There was a problem hiding this comment.
The docstring states "Direct .xts file usage from cwd or as the first argument is not supported." This is a breaking change from the previous behavior where users could run 'xts file.xts command'. This should be documented in the module docstring or changelog, and error messages should clearly guide users to use aliases instead.
| except: | ||
| from xts_core import utils | ||
|
|
||
| try: | ||
| from .xts_arg_parser import XTSArgumentParser | ||
| except: |
There was a problem hiding this comment.
Except block directly handles BaseException.
| except: | |
| from xts_core import utils | |
| try: | |
| from .xts_arg_parser import XTSArgumentParser | |
| except: | |
| except ImportError: | |
| from xts_core import utils | |
| try: | |
| from .xts_arg_parser import XTSArgumentParser | |
| except ImportError: |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 2 out of 2 changed files in this pull request and generated 17 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| str | None: Local path if found, else None. | ||
| """ | ||
| aliases = load_aliases() | ||
| return aliases.get(alias_name) | ||
|
|
||
|
|
There was a problem hiding this comment.
The function never validates that the resolved path actually exists or is a valid .xts file. If the aliases.json file contains stale entries pointing to deleted files or corrupted data, users will get confusing errors later when the system tries to load the config. Consider adding validation that the path exists and is readable, or at least document this as the caller's responsibility.
| str | None: Local path if found, else None. | |
| """ | |
| aliases = load_aliases() | |
| return aliases.get(alias_name) | |
| str | None: Local path if found and valid, else None. | |
| """ | |
| aliases = load_aliases() | |
| resolved = aliases.get(alias_name) | |
| if resolved is None: | |
| return None | |
| try: | |
| p = Path(resolved).expanduser() | |
| if not p.is_file(): | |
| utils.warning(f"Alias '{alias_name}' points to missing file: {p}") | |
| return None | |
| if p.suffix.lower() != ".xts": | |
| utils.warning(f"Alias '{alias_name}' does not point to a .xts file: {p}") | |
| return None | |
| # Return normalized string path | |
| return str(p) | |
| except Exception: | |
| # If anything goes wrong during validation, fail gracefully | |
| utils.warning(f"Failed to validate alias '{alias_name}' mapped to: {resolved}") | |
| return None |
| def add_alias_from_input(input_value: str, name: str | None) -> list[tuple[str, str]]: | ||
| """ | ||
| Add alias(es) from a URL, file path, or directory. | ||
|
|
||
| Returns: | ||
| list[tuple[str, str]]: List of (alias_name, resolved_path) entries added. |
There was a problem hiding this comment.
The docstring states the function returns list[tuple[str, str]] but the type hint uses the newer Python 3.9+ syntax. Similarly, str | None for the name parameter uses Python 3.10+ syntax. Ensure consistency with the minimum supported Python version. If targeting older Python versions, use List[Tuple[str, str]] from typing module and Optional[str] instead.
| alias_name = remaining_args[0] | ||
| resolved_xts_path = xts_alias.resolve_alias_to_xts_path(alias_name) | ||
|
|
||
| if resolved_xts_path is None: | ||
| error( | ||
| f'Unknown alias "{alias_name}". ' | ||
| 'Use "xts --alias --list" to see available aliases.' | ||
| ) | ||
| raise SystemExit(1) |
There was a problem hiding this comment.
The new implementation requires all XTS files to be accessed via aliases. The error message states 'Unknown alias "{alias_name}"', but there's no fallback to check if the user provided a direct path to an .xts file (either in the current directory or an absolute/relative path). This is a breaking change from the old behavior where users could run xts myfile.xts command directly. Users are now forced to create an alias for every .xts file they want to use, which significantly changes the user experience and breaks backward compatibility. Consider adding a fallback that checks if the argument is a valid file path before treating it as an unknown alias.
| def list_aliases() -> None: | ||
| aliases = _get_aliases() | ||
| if not aliases: | ||
| utils.warning('No aliases added') |
There was a problem hiding this comment.
When list_aliases() encounters no aliases and prints a warning, it still attempts to iterate over the empty dictionary with the for k, v in sorted(aliases.items()) loop. While this works (the loop simply doesn't execute), it's more readable to return early after printing the warning to make the control flow explicit.
| utils.warning('No aliases added') | |
| utils.warning('No aliases added') | |
| return |
| print(f"Fetching remote .xts config from: {url}") | ||
| r = requests.get(url) | ||
| r.raise_for_status() | ||
|
|
||
| with open(cached_path, "w", encoding="utf-8") as f: | ||
| f.write(r.text) |
There was a problem hiding this comment.
The URL fetching logic doesn't validate the content type or file size before writing to disk. A malicious URL could serve extremely large files (potentially filling up disk space) or serve non-.xts content that could cause issues. Consider adding: (1) a Content-Length check to limit maximum file size, (2) a Content-Type validation to ensure it's text-based content, and (3) potentially streaming the download with a size limit rather than loading the entire response into memory with r.text.
| ensure_dirs() | ||
| filename = hashlib.sha256(url.encode()).hexdigest() + ".xts" | ||
| path = os.path.join(CACHE_DIR, filename) | ||
|
|
||
| if not os.path.exists(path): | ||
| print(f"Fetching remote .xts config from: {url}") | ||
| r = requests.get(url) | ||
| r.raise_for_status() | ||
| with open(path, "w", encoding="utf-8") as f: | ||
| f.write(r.text) | ||
| # filename = hashlib.sha256(url.encode()).hexdigest() + ".xts" | ||
| filename = Path(urlparse(url).path).name | ||
| if not filename: | ||
| raise ValueError(f"URL does not contain a filename: {url}") | ||
|
|
||
| cached_path = os.path.join(CACHE_DIR, filename) | ||
|
|
||
| print(f"Fetching remote .xts config from: {url}") | ||
| r = requests.get(url) | ||
| r.raise_for_status() | ||
|
|
||
| with open(cached_path, "w", encoding="utf-8") as f: | ||
| f.write(r.text) | ||
|
|
||
| return path | ||
| return cached_path |
There was a problem hiding this comment.
The fetch_url_to_cache function now always re-downloads the URL content even if it was previously cached. The old implementation had a check if not os.path.exists(path): before fetching, which avoided unnecessary network requests. The new implementation removes this optimization and will always make a network request. This could cause performance issues and unnecessary bandwidth usage, especially for frequently accessed URLs. Consider adding back the caching logic or documenting why re-downloading is intentional.
| def run_alias_builtin(argv: list[str]) -> int: | ||
| """ | ||
| Built-in alias CLI. | ||
|
|
||
| Supported: | ||
| xts --alias --list | ||
| xts --alias --help | ||
| xts --alias --remove <name> | ||
| xts --alias --add <path|url|dir> [--name <name>] | ||
|
|
||
| Notes: | ||
| - For directories, all *.xts files are added. | ||
| """ | ||
| alias_parser = XTSArgumentParser(prog='xts --alias', add_help=True) | ||
| alias_parser.add_argument('uri', | ||
| action='store', | ||
| default=None, | ||
| help='URI of xts file to add or alias name', | ||
| nargs='?') | ||
| alias_parser.add_argument('--list', | ||
| action='store_true', | ||
| default=False, | ||
| help='List all aliases') | ||
| alias_parser.add_argument('--remove', '--rm', | ||
| action='store', | ||
| help='Remove alias', | ||
| default=None, | ||
| metavar='ALIAS_NAME', | ||
| dest='remove') | ||
| alias_parser.add_argument('--add', | ||
| action='store', | ||
| metavar='URI', | ||
| default=None, | ||
| help='Add an alias of an xts file URI') | ||
| alias_parser.add_argument('--name', | ||
| action='store', | ||
| help='Name to use for alias') | ||
| # show help & current aliases if no args | ||
| if not argv: | ||
| alias_parser.print_help() | ||
| print("\nCurrent aliases:") | ||
| list_aliases() | ||
| return 0 | ||
|
|
||
| args = alias_parser.parse_args(argv) | ||
|
|
||
| if args.list: | ||
| list_aliases() | ||
| return 0 | ||
|
|
||
| if args.remove is not None: | ||
| if remove_alias(args.remove): | ||
| print(f"Removed alias: {args.remove}") | ||
| return 0 | ||
| print(f"Alias not found: {args.remove}") | ||
| return 2 | ||
|
|
||
| if args.uri and args.add: | ||
| alias_parser.error('Provide the URI only once (either positional OR via "--add").') | ||
| return 2 | ||
|
|
||
| input_value = args.add or args.uri | ||
| if not input_value: | ||
| alias_parser.print_help() | ||
| return 2 | ||
|
|
||
| try: | ||
| added = add_alias_from_input(input_value, args.name) | ||
| except (FileNotFoundError, ValueError) as e: | ||
| alias_parser.error(str(e)) | ||
| return 2 | ||
| except Exception as e: | ||
| utils.error(f"Failed to add alias: {str(e)}") | ||
| return 2 | ||
|
|
||
| for k, v in added: | ||
| print(f"{k} -> {v}") | ||
| return 0 |
There was a problem hiding this comment.
The new alias functionality including run_alias_builtin, add_alias_from_input, cache_local_file_to_cache, resolve_alias_to_xts_path, and remove_alias lack test coverage. The repository has comprehensive tests for the XTSAllocatorClient plugin, but no tests exist for the xts_alias module functionality. Given that this is a significant rework of the alias system with complex file caching logic and multiple code paths (URL/file/directory), these functions should have corresponding tests to ensure correctness and prevent regressions.
| try: | ||
| cache_root = Path(CACHE_DIR).expanduser().resolve() | ||
| target = Path(cached_path).expanduser().resolve() | ||
| if cache_root in target.parents and target.is_file(): |
There was a problem hiding this comment.
The cache checking logic has a bug. The condition cache_root in target.parents checks if cache_root is in the list of parent directories, but this should be checking if the target is a descendant of cache_root. The correct check should be target.is_relative_to(cache_root) (Python 3.9+) or checking if the path starts with the cache directory. The current implementation will fail because cache_root would need to be exactly equal to one of the parent directories, not checking if target is contained within the cache hierarchy.
| if cache_root in target.parents and target.is_file(): | |
| # Ensure we only delete files that are descendants of the cache root | |
| try: | |
| is_in_cache = target.is_relative_to(cache_root) # Python 3.9+ | |
| except AttributeError: | |
| # Fallback for Python < 3.9 | |
| try: | |
| is_in_cache = os.path.commonpath([str(cache_root), str(target)]) == str(cache_root) | |
| except ValueError: | |
| # Occurs if paths are on different drives on Windows, etc. | |
| is_in_cache = False | |
| if is_in_cache and target.is_file(): |
| - `xts --alias --remove <name>` removes the alias mapping from aliases.json | ||
| """ | ||
|
|
||
| import argparse |
There was a problem hiding this comment.
The import of argparse is unused. The file only uses XTSArgumentParser from the xts_arg_parser module. This unnecessary import should be removed to keep the code clean.
| import argparse |
3 participants
No description provided.