Warning

This document is for an in-development version of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for cwltool.stdfsaccess

"""Abstracted IO access."""
from __future__ import absolute_import

import glob
import os
from io import open
from typing import IO, Any, List

from schema_salad.ref_resolver import file_uri, uri_file_path
from six.moves import urllib
from typing_extensions import Text
# move to a regular typing import when Python 3.3-3.6 is no longer supported

from .utils import onWindows


def abspath(src, basedir):  # type: (Text, Text) -> Text
    if src.startswith(u"file://"):
        abpath = Text(uri_file_path(str(src)))
    elif urllib.parse.urlsplit(src).scheme in ['http', 'https']:
        return src
    else:
        if basedir.startswith(u"file://"):
            abpath = src if os.path.isabs(src) else basedir+ '/'+ src
        else:
            abpath = src if os.path.isabs(src) else os.path.join(basedir, src)
    return abpath

[docs]class StdFsAccess(object): """Local filesystem implementation."""
[docs] def __init__(self, basedir): # type: (Text) -> None """Perform operations with respect to a base directory.""" self.basedir = basedir
def _abs(self, p): # type: (Text) -> Text return abspath(p, self.basedir)
[docs] def glob(self, pattern): # type: (Text) -> List[Text] return [file_uri(str(self._abs(l))) for l in glob.glob(self._abs(pattern))]
[docs] def open(self, fn, mode): # type: (Text, str) -> IO[Any] return open(self._abs(fn), mode)
[docs] def exists(self, fn): # type: (Text) -> bool return os.path.exists(self._abs(fn))
[docs] def size(self, fn): # type: (Text) -> int return os.stat(self._abs(fn)).st_size
[docs] def isfile(self, fn): # type: (Text) -> bool return os.path.isfile(self._abs(fn))
[docs] def isdir(self, fn): # type: (Text) -> bool return os.path.isdir(self._abs(fn))
[docs] def listdir(self, fn): # type: (Text) -> List[Text] return [abspath(urllib.parse.quote(str(l)), fn) for l in os.listdir(self._abs(fn))]
[docs] def join(self, path, *paths): # type: (Text, *Text) -> Text return os.path.join(path, *paths)
[docs] def realpath(self, path): # type: (Text) -> Text return os.path.realpath(path)
# On windows os.path.realpath appends unecessary Drive, here we would avoid that
[docs] def docker_compatible_realpath(self, path): # type: (Text) -> Text if onWindows(): if path.startswith('/'): return path return '/'+path return self.realpath(path)