Skip to content

Commit

Permalink
path_to_str -> str
Browse files Browse the repository at this point in the history
  • Loading branch information
khsrali committed Jan 20, 2025
1 parent 91ea017 commit b22338e
Show file tree
Hide file tree
Showing 4 changed files with 122 additions and 129 deletions.
84 changes: 42 additions & 42 deletions src/aiida/transports/plugins/local.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@

from aiida.common.warnings import warn_deprecation
from aiida.transports import cli as transport_cli
from aiida.transports.transport import BlockingTransport, TransportInternalError, TransportPath, path_to_str
from aiida.transports.transport import BlockingTransport, TransportInternalError, TransportPath


# refactor or raise the limit: issue #1784
Expand Down Expand Up @@ -107,7 +107,7 @@ def chdir(self, path: TransportPath):
'`chdir()` is deprecated and will be removed in the next major version.',
version=3,
)
path = path_to_str(path)
path = str(path)
new_path = os.path.join(self.curdir, path)
if not os.path.isdir(new_path):
raise OSError(f"'{new_path}' is not a valid directory")
Expand All @@ -117,14 +117,14 @@ def chdir(self, path: TransportPath):
self._internal_dir = os.path.normpath(new_path)

def chown(self, path: TransportPath, uid, gid):
path = path_to_str(path)
path = str(path)

Check warning on line 120 in src/aiida/transports/plugins/local.py

View check run for this annotation

Codecov / codecov/patch

src/aiida/transports/plugins/local.py#L120

Added line #L120 was not covered by tests
os.chown(path, uid, gid)

def normalize(self, path: TransportPath = '.'):
"""Normalizes path, eliminating double slashes, etc..
:param path: path to normalize
"""
path = path_to_str(path)
path = str(path)
return os.path.realpath(os.path.join(self.curdir, path))

def getcwd(self):
Expand All @@ -138,7 +138,7 @@ def getcwd(self):
@staticmethod
def _os_path_split_asunder(path: TransportPath):
"""Used by makedirs, Takes path (a str) and returns a list deconcatenating the path."""
path = path_to_str(path)
path = str(path)
parts = []
while True:
newpath, tail = os.path.split(path)
Expand All @@ -163,7 +163,7 @@ def makedirs(self, path: TransportPath, ignore_existing=False):
:raise OSError: If the directory already exists and is not ignore_existing
"""
path = path_to_str(path)
path = str(path)
# check to avoid creation of empty dirs
path = os.path.normpath(path)

Expand All @@ -188,7 +188,7 @@ def mkdir(self, path: TransportPath, ignore_existing=False):
:raise OSError: If the directory already exists.
"""
path = path_to_str(path)
path = str(path)
if ignore_existing and self.isdir(path):
return

Expand All @@ -198,14 +198,14 @@ def rmdir(self, path: TransportPath):
"""Removes a folder at location path.
:param path: path to remove
"""
path = path_to_str(path)
path = str(path)
os.rmdir(os.path.join(self.curdir, path))

def isdir(self, path: TransportPath):
"""Checks if 'path' is a directory.
:return: a boolean
"""
path = path_to_str(path)
path = str(path)
if not path:
return False

Expand All @@ -218,7 +218,7 @@ def chmod(self, path: TransportPath, mode):
:raise OSError: if path does not exist.
"""
path = path_to_str(path)
path = str(path)
if not path:
raise OSError('Directory not given in input')
real_path = os.path.join(self.curdir, path)
Expand All @@ -243,8 +243,8 @@ def put(self, localpath: TransportPath, remotepath: TransportPath, *args, **kwar
:raise OSError: if remotepath is not valid
:raise ValueError: if localpath is not valid
"""
localpath = path_to_str(localpath)
remotepath = path_to_str(remotepath)
localpath = str(localpath)
remotepath = str(remotepath)
from aiida.common.warnings import warn_deprecation

if 'ignore_noexisting' in kwargs:
Expand Down Expand Up @@ -324,8 +324,8 @@ def putfile(self, localpath: TransportPath, remotepath: TransportPath, *args, **
:raise ValueError: if localpath is not valid
:raise OSError: if localpath does not exist
"""
localpath = path_to_str(localpath)
remotepath = path_to_str(remotepath)
localpath = str(localpath)
remotepath = str(remotepath)

overwrite = kwargs.get('overwrite', args[0] if args else True)
if not remotepath:
Expand Down Expand Up @@ -360,8 +360,8 @@ def puttree(self, localpath: TransportPath, remotepath: TransportPath, *args, **
:raise ValueError: if localpath is not valid
:raise OSError: if localpath does not exist
"""
localpath = path_to_str(localpath)
remotepath = path_to_str(remotepath)
localpath = str(localpath)
remotepath = str(remotepath)
dereference = kwargs.get('dereference', args[0] if args else True)
overwrite = kwargs.get('overwrite', args[1] if len(args) > 1 else True)
if not remotepath:
Expand Down Expand Up @@ -392,7 +392,7 @@ def rmtree(self, path: TransportPath):
:param path: a string to path
"""
path = path_to_str(path)
path = str(path)
the_path = os.path.join(self.curdir, path)
try:
shutil.rmtree(the_path)
Expand Down Expand Up @@ -421,8 +421,8 @@ def get(self, remotepath: TransportPath, localpath: TransportPath, *args, **kwar
:raise OSError: if 'remote' remotepath is not valid
:raise ValueError: if 'local' localpath is not valid
"""
remotepath = path_to_str(remotepath)
localpath = path_to_str(localpath)
remotepath = str(remotepath)
localpath = str(localpath)
dereference = kwargs.get('dereference', args[0] if args else True)
overwrite = kwargs.get('overwrite', args[1] if len(args) > 1 else True)
ignore_nonexisting = kwargs.get('ignore_nonexisting', args[2] if len(args) > 2 else False)
Expand Down Expand Up @@ -487,8 +487,8 @@ def getfile(self, remotepath: TransportPath, localpath: TransportPath, *args, **
:raise ValueError: if 'local' localpath is not valid
:raise OSError: if unintentionally overwriting
"""
remotepath = path_to_str(remotepath)
localpath = path_to_str(localpath)
remotepath = str(remotepath)
localpath = str(localpath)

if not os.path.isabs(localpath):
raise ValueError('localpath must be an absolute path')
Expand Down Expand Up @@ -521,8 +521,8 @@ def gettree(self, remotepath: TransportPath, localpath: TransportPath, *args, **
:raise ValueError: if 'local' localpath is not valid
:raise OSError: if unintentionally overwriting
"""
remotepath = path_to_str(remotepath)
localpath = path_to_str(localpath)
remotepath = str(remotepath)
localpath = str(localpath)
dereference = kwargs.get('dereference', args[0] if args else True)
overwrite = kwargs.get('overwrite', args[1] if len(args) > 1 else True)
if not remotepath:
Expand Down Expand Up @@ -562,8 +562,8 @@ def copy(self, remotesource: TransportPath, remotedestination: TransportPath, de
:raise ValueError: if 'remote' remotesource or remotedestinationis not valid
:raise OSError: if remotesource does not exist
"""
remotesource = path_to_str(remotesource)
remotedestination = path_to_str(remotedestination)
remotesource = str(remotesource)
remotedestination = str(remotedestination)
if not remotesource:
raise ValueError('Input remotesource to copy must be a non empty object')
if not remotedestination:
Expand Down Expand Up @@ -622,8 +622,8 @@ def copyfile(self, remotesource: TransportPath, remotedestination: TransportPath
:raise ValueError: if 'remote' remotesource or remotedestination is not valid
:raise OSError: if remotesource does not exist
"""
remotesource = path_to_str(remotesource)
remotedestination = path_to_str(remotedestination)
remotesource = str(remotesource)
remotedestination = str(remotedestination)

Check warning on line 626 in src/aiida/transports/plugins/local.py

View check run for this annotation

Codecov / codecov/patch

src/aiida/transports/plugins/local.py#L625-L626

Added lines #L625 - L626 were not covered by tests
if not remotesource:
raise ValueError('Input remotesource to copyfile must be a non empty object')
if not remotedestination:
Expand All @@ -650,8 +650,8 @@ def copytree(self, remotesource: TransportPath, remotedestination: TransportPath
:raise ValueError: if 'remote' remotesource or remotedestination is not valid
:raise OSError: if remotesource does not exist
"""
remotesource = path_to_str(remotesource)
remotedestination = path_to_str(remotedestination)
remotesource = str(remotesource)
remotedestination = str(remotedestination)
if not remotesource:
raise ValueError('Input remotesource to copytree must be a non empty object')
if not remotedestination:
Expand All @@ -672,7 +672,7 @@ def get_attribute(self, path: TransportPath):
as specified in aiida.transports.
:param path: the path of the given file.
"""
path = path_to_str(path)
path = str(path)
from aiida.transports.util import FileAttribute

os_attr = os.lstat(os.path.join(self.curdir, path))
Expand All @@ -687,7 +687,7 @@ def _local_listdir(self, path: TransportPath, pattern=None):
"""Act on the local folder, for the rest, same as listdir."""
import re

path = path_to_str(path)
path = str(path)

Check warning on line 690 in src/aiida/transports/plugins/local.py

View check run for this annotation

Codecov / codecov/patch

src/aiida/transports/plugins/local.py#L690

Added line #L690 was not covered by tests

if not pattern:
return os.listdir(path)
Expand All @@ -708,7 +708,7 @@ def listdir(self, path: TransportPath = '.', pattern=None):
:param pattern: if set, returns the list of files matching pattern.
Unix only. (Use to emulate ls * for example)
"""
path = path_to_str(path)
path = str(path)
the_path = os.path.join(self.curdir, path).strip()
if not pattern:
try:
Expand All @@ -727,14 +727,14 @@ def listdir(self, path: TransportPath = '.', pattern=None):

def remove(self, path: TransportPath):
"""Removes a file at position path."""
path = path_to_str(path)
path = str(path)
os.remove(os.path.join(self.curdir, path))

def isfile(self, path: TransportPath):
"""Checks if object at path is a file.
Returns a boolean.
"""
path = path_to_str(path)
path = str(path)
if not path:
return False
return os.path.isfile(os.path.join(self.curdir, path))
Expand Down Expand Up @@ -766,7 +766,7 @@ def _exec_command_internal(self, command, workdir: Optional[TransportPath] = Non
from aiida.common.escaping import escape_for_bash

if workdir:
workdir = path_to_str(workdir)
workdir = str(workdir)
# Note: The outer shell will eat one level of escaping, while
# 'bash -l -c ...' will eat another. Thus, we need to escape again.
bash_commmand = f'{self._bash_command_str}-c '
Expand Down Expand Up @@ -801,7 +801,7 @@ def exec_command_wait_bytes(self, command, stdin=None, workdir: Optional[Transpo
are both bytes and the return_value is an int.
"""
if workdir:
workdir = path_to_str(workdir)
workdir = str(workdir)
with self._exec_command_internal(command, workdir) as process:
if stdin is not None:
# Implicitly assume that the desired encoding is 'utf-8' if I receive a string.
Expand Down Expand Up @@ -855,7 +855,7 @@ def gotocomputer_command(self, remotedir: TransportPath):
:param str remotedir: the full path of the remote directory
"""
remotedir = path_to_str(remotedir)
remotedir = str(remotedir)
connect_string = self._gotocomputer_string(remotedir)
cmd = f'bash -c {connect_string}'
return cmd
Expand All @@ -869,8 +869,8 @@ def rename(self, oldpath: TransportPath, newpath: TransportPath):
:raises OSError: if src/dst is not found
:raises ValueError: if src/dst is not a valid string
"""
oldpath = path_to_str(oldpath)
newpath = path_to_str(newpath)
oldpath = str(oldpath)
newpath = str(newpath)

Check warning on line 873 in src/aiida/transports/plugins/local.py

View check run for this annotation

Codecov / codecov/patch

src/aiida/transports/plugins/local.py#L872-L873

Added lines #L872 - L873 were not covered by tests
if not oldpath:
raise ValueError(f'Source {oldpath} is not a valid string')
if not newpath:
Expand All @@ -889,8 +889,8 @@ def symlink(self, remotesource: TransportPath, remotedestination: TransportPath)
:param remotesource: remote source. Can contain a pattern.
:param remotedestination: remote destination
"""
remotesource = os.path.normpath(path_to_str(remotesource))
remotedestination = os.path.normpath(path_to_str(remotedestination))
remotesource = os.path.normpath(str(remotesource))
remotedestination = os.path.normpath(str(remotedestination))

if self.has_magic(remotesource):
if self.has_magic(remotedestination):
Expand All @@ -911,7 +911,7 @@ def symlink(self, remotesource: TransportPath, remotedestination: TransportPath)

def path_exists(self, path: TransportPath):
"""Check if path exists"""
path = path_to_str(path)
path = str(path)
return os.path.exists(os.path.join(self.curdir, path))


Expand Down
Loading

0 comments on commit b22338e

Please sign in to comment.