Fix #261 - pkg_resources migration

chanjo/__init__.py

@@ -8,7 +8,11 @@
 :licence: MIT, see LICENCE for more details
 """
 import logging
-from pkg_resources import get_distribution
+try:
+    from importlib.metadata import version
+except ImportError:  # Backport support for importlib metadata on Python 3.7
+    from importlib_metadata import version
+
 
 __banner__ = r"""
        ______               ________
@@ -23,7 +27,7 @@
 __summary__ = 'coverage analysis tool for clinical sequencing'
 __uri__ = 'http://www.chanjo.co/'
 
-__version__ = get_distribution(__title__).version
+__version__ = version(__title__)
 __codename__ = 'Optimistic Otter'
 
 __author__ = 'Robin Andeer'

chanjo/cli/base.py

@@ -6,7 +6,10 @@
 """
 import logging
 import os
-import pkg_resources
+try:
+    from importlib.metadata import entry_points
+except ImportError:  # Backport support for importlib metadata on Python 3.7
+    from importlib_metadata import entry_points
 
 import click
 import coloredlogs
@@ -16,14 +19,20 @@
 
 LOG = logging.getLogger(__name__)
 
+COMMAND_GROUP_KEY = 'chanjo.subcommands.4'
 
 class EntryPointsCLI(click.MultiCommand):
     """Add sub-commands dynamically to a CLI via entry points."""
 
     def _iter_commands(self):
         """Iterate over all sub-commands as defined by the entry point."""
-        return {entry_point.name: entry_point for entry_point in
-                pkg_resources.iter_entry_points('chanjo.subcommands.4')}
+        # Get all entry points for the specified group
+        if hasattr(entry_points(), 'select'):
+            # Python >3.10, importlib
+            eps = entry_points(group=COMMAND_GROUP_KEY)
+        else:
+            eps = entry_points().get(COMMAND_GROUP_KEY, [])
+        return {ep.name: ep for ep in eps}
 
     def list_commands(self, ctx):
         """List the available commands."""
@@ -34,7 +43,7 @@ def get_command(self, ctx, name):
         """Load one of the available commands."""
         commands = self._iter_commands()
         if name not in commands:
-            click.echo("no such command: {}".format(name))
+            click.echo(f"no such command: {name}")
             ctx.abort()
         return commands[name].load()
 
@@ -51,15 +60,15 @@ def root(context, config, database, log_level, log_file):
     """Clinical sequencing coverage analysis tool."""
     logout = log_file or click.get_text_stream('stderr')
     coloredlogs.install(level=log_level, stream=logout)
-    LOG.debug("version {0}".format(__version__))
+    LOG.debug("version %s", __version__)
 
-    # avoid setting global defaults in Click options, do it below when
+    # Load configuration from the provided file
     if os.path.exists(config):
         with open(config) as conf_handle:
             context.obj = yaml.safe_load(conf_handle)
     else:
         context.obj = {}
     context.obj['database'] = (database or context.obj.get('database'))
 
-    # update the context with new defaults from the config file
+    # Update the context with new defaults from the config file
     context.default_map = context.obj

chanjo/cli/init.py

@@ -1,10 +1,10 @@
 # -*- coding: utf-8 -*-
 import codecs
-from distutils.spawn import find_executable
+from shutil import which
 import logging
 
 import click
-from path import Path
+from pathlib import Path
 import yaml
 
 from chanjo.store.api import ChanjoDB
@@ -27,10 +27,10 @@ def init(context, force, demo, auto, root_dir):
 
     LOG.info("setting up chanjo under: %s", root_path)
     db_uri = context.obj.get('database')
-    db_uri = db_uri or "sqlite:///{}".format(root_path.joinpath(DB_NAME).abspath())
+    db_uri = db_uri or "sqlite:///{}".format(root_path.joinpath(DB_NAME).resolve())
 
     # test setup of sambamba
-    sambamba_bin = find_executable('sambamba')
+    sambamba_bin = which('sambamba')
     if sambamba_bin is None:  # pragma: no cover
         LOG.warning("'sambamba' command not found")
     else:
@@ -53,7 +53,7 @@ def init(context, force, demo, auto, root_dir):
         is_bootstrapped = True
 
     # setup config file
-    root_path.makedirs_p()
+    root_path.mkdir(parents=True, exist_ok=True)
     conf_path = root_path.joinpath('chanjo.yaml')
     with open(conf_path, 'w') as conf_handle:
         data = {'database': db_uri}

chanjo/cli/load.py

@@ -1,8 +1,9 @@
 # -*- coding: utf-8 -*-
-import os.path
 import logging
 import sys
 
+from pathlib import Path
+
 import click
 from sqlalchemy.exc import IntegrityError
 
@@ -12,7 +13,6 @@
 
 LOG = logging.getLogger(__name__)
 
-
 def validate_stdin(context, param, value):
     """Validate piped input contains some data.
 
@@ -39,7 +39,7 @@ def validate_stdin(context, param, value):
 def load(context, sample, group, name, group_name, threshold, bed_stream):
     """Load Sambamba output into the database for a sample."""
     chanjo_db = ChanjoDB(uri=context.obj['database'])
-    source = os.path.abspath(bed_stream.name)
+    source = str(Path(bed_stream.name).resolve())
 
     result = load_transcripts(bed_stream, sample_id=sample, group_id=group,
                               source=source, threshold=threshold)

chanjo/init/bootstrap.py

@@ -3,12 +3,14 @@
 import zipfile
 import sys
 
+from pathlib import Path
+
 if sys.version_info[0] >=3:
     from urllib.request import urlretrieve
 else:
     from urllib import urlretrieve
 
-from path import Path
+
 
 DB_NAME = 'chanjo.coverage.sqlite3'
 BED_NAME = 'hgnc.grch37p13.exons.bed'
@@ -26,7 +28,7 @@ def pull(target_dir, force=False):  # pragma: no cover
     """
     logger.debug('ensure target directory exists')
     target_path = Path(target_dir)
-    target_path.makedirs_p()
+    target_path.mkdir(parents=True, exist_ok=True)
 
     bed_zip_path = target_path.joinpath("{}.zip".format(BED_NAME))
     final_bed = target_path.joinpath(BED_NAME)
@@ -40,6 +42,6 @@ def pull(target_dir, force=False):  # pragma: no cover
         zip_ref.extractall(target_dir)
 
         logger.info('removing BED archive...')
-        bed_zip_path.remove_p()
+        bed_zip_path.unlink()
     else:
         logger.warn('file already exists, skipping: %s', final_bed)

chanjo/init/demo.py

@@ -1,37 +1,49 @@
 # -*- coding: utf-8 -*-
 from errno import EEXIST
 import logging
-from pkg_resources import resource_filename, resource_listdir
+import shutil
+from pathlib import Path
+
+try:
+    from importlib.resources import files
+except ImportError:  # Backport support for importlib metadata on Python 3.7
+    from importlib_resources import files
 
-from path import Path
 
 DEMO_BED_NAME = 'hgnc.min.bed'
 log = logging.getLogger(__name__)
 
-
 def setup_demo(location, force=False):
     """Copy demo files to a directory.
 
-    \b
-    LOCATION: directory to add demofiles to (default: ./chanjo-demo)
+    LOCATION: directory to add demo files to (default: ./chanjo-demo)
     """
     target_dir = Path(location)
     pkg_dir = __name__.rpartition('.')[0]
-    demo_dir = Path(resource_filename(pkg_dir, 'demo-files'))
 
-    # make sure we don't overwrite exiting files
-    for demo_file in resource_listdir(pkg_dir, 'demo-files'):
-        target_file_path = target_dir.joinpath(demo_file)
+    # Get the demo-files directory path using importlib.resources
+    demo_dir = files(pkg_dir) / 'demo-files'
+
+    if not demo_dir.is_dir():
+        log.error("Demo files directory does not exist")
+        raise FileNotFoundError(f"'demo-files' directory not found in package {pkg_dir}")
+
+    # Check for existing files and avoid overwriting unless `force` is True
+    for demo_file in demo_dir.iterdir():
+        target_file_path = target_dir / demo_file.name
         if not force and target_file_path.exists():
             log.error("%s exists, pick a different location", target_file_path)
             raise OSError(EEXIST, 'file already exists', target_file_path)
 
     try:
-        # we can copy the directory(tree)
-        demo_dir.copytree(target_dir)
+        # Copy the directory tree
+        shutil.copytree(demo_dir, target_dir)
+    except FileExistsError:
+        log.warning('Location must be a non-existing directory')
+        raise
     except OSError as error:
-        log.warn('location must be a non-existing directory')
-        raise error
+        log.error('An error occurred during file copying: %s', error)
+        raise
 
-    # inform the user
-    log.info("successfully copied demo files to %s", target_dir)
+    # Inform the user
+    log.info("Successfully copied demo files to %s", target_dir)

chanjo/store/api.py

@@ -3,6 +3,8 @@
 import logging
 import os
 
+from pathlib import Path
+
 from sqlservice import Database
 from chanjo.calculate import CalculateMixin
 from .models import BASE
@@ -11,6 +13,11 @@
 
 LOG = logging.getLogger(__name__)
 
+def get_absolute_path(db_uri):
+    """Get the absolute path of the database URI."""
+    # Use Path for handling paths
+    db_path = Path(db_uri).expanduser().resolve()
+    return db_path
 
 class ChanjoDB(Database, CalculateMixin, DeleteMixin, FetchMixin):
     """SQLAlchemy-based database object.
@@ -62,7 +69,7 @@ def connect(self, db_uri, debug=False):
             config['SQLALCHEMY_POOL_RECYCLE'] = 3600
         elif '://' not in db_uri:
             # expect only a path to a sqlite database
-            db_path = os.path.abspath(os.path.expanduser(db_uri))
+            db_path = get_absolute_path(db_uri)
             db_uri = "sqlite:///{}".format(db_path)
 
         config['SQLALCHEMY_DATABASE_URI'] = db_uri

chanjo/testutils.py

@@ -1,5 +1,8 @@
-import os
+from pathlib import Path
 
+import logging
+
+LOG = logging.getLogger(__name__)
 
 def fake_urlretrieve(url, target):
     open(target, 'a').close()
@@ -11,5 +14,6 @@ def __init__(self, in_path, mode='r'):
         self.in_path = in_path
 
     def extractall(self, target_dir):
-        out_path = os.path.join(target_dir, self.in_path.replace('.zip', ''))
+        LOG.info(f"fake zip target dir : {target_dir} path {self.in_path}")
+        out_path = Path(target_dir).joinpath(self.in_path.name.replace('.zip', ''))
         open(out_path, 'a').close()

requirements.txt

@@ -1,10 +1,10 @@
 coloredlogs
 click
 cryptography
-path.py
 toolz
 pyyaml
 importlib_metadata
+importlib_resources
 pymysql
 sqlalchemy>=2.0
 sqlservice>=3.0

tests/cli/test_cli_base.py

@@ -4,14 +4,14 @@
 import yaml
 
 
-def test_logging_to_file(tmpdir, invoke_cli):
+def test_logging_to_file(tmp_path, invoke_cli):
     # GIVEN an empty directory
-    assert tmpdir.listdir() == []
+    assert not list(tmp_path.iterdir())
     # WHEN running the CLI to display some help for a subcommand
-    log_path = tmpdir.join('stderr.log')
+    log_path = tmp_path.joinpath('stderr.log')
     result = invoke_cli(['--log-file', str(log_path), 'db'])
     assert result.exit_code == 0
-    assert tmpdir.listdir() == [log_path]
+    assert list(tmp_path.iterdir()) == [log_path]
 
 
 def test_list_commands(invoke_cli):

tests/cli/test_cli_init.py

@@ -7,40 +7,40 @@
 from chanjo.testutils import FakeZipFile, fake_urlretrieve
 
 
-def test_init_demo(tmpdir, invoke_cli):
+def test_init_demo(tmp_path, invoke_cli):
     # GIVEN empty directory
-    assert tmpdir.listdir() == []
+    assert not list(tmp_path.iterdir())
     # WHEN setting up demo on CLI
-    target_dir = tmpdir.join('chanjo-demo')
+    target_dir = tmp_path.joinpath('chanjo-demo')
     target_path = str(target_dir)
     result = invoke_cli(['init', '--demo', target_path])
     # THEN is should work and place 4 demo files + 1 sqlite db + 1 config
     assert result.exit_code == 0
-    assert len(target_dir.listdir()) == (4 + 1 + 1)
+    assert len(list(target_dir.iterdir())) == (4 + 1 + 1)
 
 
 @patch('urllib.request.urlretrieve', fake_urlretrieve)
 @patch('zipfile.ZipFile', FakeZipFile)
 @patch('click.confirm', lambda param: True)
-def test_init_bootstrap(tmpdir, invoke_cli):
+def test_init_bootstrap(tmp_path, invoke_cli):
     # GIVEN empty dir
-    assert tmpdir.listdir() == []
-    # WHEN boostrapping chanjo
-    result = invoke_cli(['init', str(tmpdir)])
+    assert not list(tmp_path.iterdir())
+    # WHEN bootstrapping chanjo
+    result = invoke_cli(['init', str(tmp_path)])
     # THEN it should place 2 + 1 (BED, DB, config) files in the target dir
     assert result.exit_code == 0
-    assert len(tmpdir.listdir()) == 3
+    assert len(list(tmp_path.iterdir())) == 3
 
 
 @patch('click.confirm', lambda param: False)
 @patch('click.prompt', lambda param: "{}/cov.sqlite3".format(gettempdir()))
-def test_init_no_bootstrap(tmpdir, invoke_cli):
+def test_init_no_bootstrap(tmp_path, invoke_cli):
     # GIVEN ...
     # WHEN opting out of bootstrap
-    result = invoke_cli(['init', str(tmpdir)])
+    result = invoke_cli(['init', str(tmp_path)])
     # THEN it should work with custom database URI
     assert result.exit_code == 0
-    conf_path = tmpdir.join('chanjo.yaml')
+    conf_path = tmp_path.joinpath('chanjo.yaml')
     with open(str(conf_path), 'r') as handle:
         data = yaml.safe_load(handle)
     assert 'coverage.sqlite3' in data['database']