Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@require_connection
def is_port_bound(self, host, port):
"""
Check whether a port on a remote host is accessible.
This method checks to see whether a particular port is active on a
given host by attempting to establish a connection with it.
Args:
host (str): The hostname of the target service.
port (int): The port of the target service.
Returns:
bool: Whether the port is active and accepting connections.
"""
return self._is_port_bound(host, port)
@require_connection
def showdir(self, path=None):
"""
Return a dataframe representation of a directory.
This method returns a `pandas.DataFrame` representation of the contents of
a path, which are retrieved using `.dir(path)`. The exact columns will
vary from filesystem to filesystem, depending on the fields returned
by `.dir()`, but the returned DataFrame is guaranteed to at least have
the columns: 'name' and 'type'.
Args:
path (str): The path of the directory from which to show contents.
Returns:
pandas.DataFrame: A DataFrame representation of the contents of the
nominated directory.
@require_connection
def walk(self, path=None):
"""
Explore the filesystem tree starting at a nominated path.
This method returns a generator which recursively walks over all paths
that are children of `path`, one result for each directory, of form:
(
@require_connection
def schemas(self):
"""
object: An object with attributes corresponding to the names of the schemas
in this database.
"""
from werkzeug import LocalProxy
def get_schemas():
if not getattr(self, '_schemas', None):
assert getattr(self, '_sqlalchemy_metadata', None) is not None, (
"`{class_name}` instances do not provide the required sqlalchemy metadata "
"for schema exploration.".format(self.__class__.__name__)
)
self._schemas = Schemas(self._sqlalchemy_metadata)
return self._schemas
return LocalProxy(get_schemas)
@require_connection
def _file_read(self, path, size=-1, offset=0, binary=False):
"""
This method is used by `FileSystemFile` to read the contents of files.
`._file_read_` may be left unimplemented if `.open()` returns a different
kind of file handle.
Args:
path (str): The path of the file to be read.
size (int): The number of bytes to read at a time (-1 for max possible).
offset (int): The offset in bytes from the start of the file.
binary (bool): Whether to read the file in binary mode.
Returns:
str or bytes: The contents of the file.
"""
return self._file_read_(self._path(path), size=size, offset=offset, binary=binary)
@require_connection
def dataframe_to_table(self, df, table, if_exists='fail', **kwargs):
"""
Upload a local pandas dataframe into a table in this database.
Args:
df (pandas.DataFrame): The dataframe to upload into the database.
table (str, ParsedNamespaces): The name of the table into which the
dataframe should be uploaded.
if_exists (str): if nominated table already exists: 'fail' to do
nothing, 'replace' to drop, recreate and insert data into new
table, and 'append' to add data from this table into the
existing table.
**kwargs (dict): Additional keyword arguments to pass onto
`DatabaseClient._dataframe_to_table`.
"""
assert if_exists in {'fail', 'replace', 'append'}
@require_connection
def request(self, endpoint, method='get', **kwargs):
"""
Request data from a nominated endpoint.
Args:
endpoint (str): The endpoint from which to receive data.
method (str): The method to use when requestion this resource.
**kwargs (dict): Additional arguments to pass through to
`requests.request`.
Returns:
requests.Response: The response object associated with this request.
"""
import requests
url = urljoin(self.base_url, endpoint)
return requests.request(method, url, **kwargs)
@require_connection
def find(self, path_prefix=None, **attrs):
"""
Find a file or directory based on certain attributes.
This method searches for files or folders which satisfy certain
constraints on the attributes of the file (as encoded into
`FileSystemFileDesc`). Note that without attribute constraints,
this method will function identically to `self.dir`.
Args:
path_prefix (str): The path under which files/directories should be
found.
**attrs (dict): Constraints on the fields of the `FileSystemFileDesc`
objects associated with this filesystem, as constant values or
callable objects (in which case the object will be called and
should return True if attribute value is match, and False
@require_connection
def get_metadata(self, key, namespace=None):
"""
Retrieve metadata associated with the nominated key from the cache.
Args:
key (str): The key for which to extract metadata.
namespace (str, None): The namespace to be used.
Returns:
dict: The metadata associated with this namespace and key.
"""
namespace, key = self._namespace(namespace), self._key(key)
if not self._has_key(namespace, key):
raise KeyError("{} (namespace: {})".format(key, namespace))
try:
with self._get_stream_for_key(namespace, key, 'metadata', mode='r', create=False) as fh:
@require_connection
def exists(self, path):
"""
Check whether nominated path exists on this filesytem.
Args:
path (str): The path for which to check existence.
Returns:
bool: `True` if file/folder exists at nominated path, and `False`
otherwise.
"""
return self._exists(self._path(path))