### Install Dependencies Source: https://waterbutler.readthedocs.io/en/latest/gettingstarted.html Install the required setuptools and invoke versions for the environment. ```bash pip install setuptools==37.0.0 pip install invoke==0.13.0 ``` -------------------------------- ### Run Installation Tasks Source: https://waterbutler.readthedocs.io/en/latest/gettingstarted.html Execute the installation process for the project, with an optional development mode. ```bash invoke install ``` ```bash invoke install --develop ``` -------------------------------- ### Start Network Server Source: https://waterbutler.readthedocs.io/en/latest/_modules/asyncio/streams.html Starts a socket server that invokes a callback for each new client connection. ```python async def start_server(client_connected_cb, host=None, port=None, *, loop=None, limit=_DEFAULT_LIMIT, **kwds): """Start a socket server, call back for each client connected. The first parameter, `client_connected_cb`, takes two parameters: client_reader, client_writer. client_reader is a StreamReader object, while client_writer is a StreamWriter object. This parameter can either be a plain callback function or a coroutine; if it is a coroutine, it will be automatically converted into a Task. The rest of the arguments are all the usual arguments to loop.create_server() except protocol_factory; most common are positional host and port, with various optional keyword arguments following. The return value is the same as loop.create_server(). Additional optional keyword arguments are loop (to set the event loop instance to use) and limit (to set the buffer limit passed to the StreamReader). The return value is the same as loop.create_server(), i.e. a Server object which can be used to stop the service. """ if loop is None: loop = events.get_event_loop() def factory(): reader = StreamReader(limit=limit, loop=loop) protocol = StreamReaderProtocol(reader, client_connected_cb, loop=loop) return protocol return await loop.create_server(factory, host, port, **kwds) ``` -------------------------------- ### Start Services Source: https://waterbutler.readthedocs.io/en/latest/gettingstarted.html Commands to launch the server and the Celery worker process. ```bash invoke server ``` ```bash invoke celery ``` -------------------------------- ### Run Tests Source: https://waterbutler.readthedocs.io/en/latest/gettingstarted.html Execute the test suite after ensuring development requirements are installed. ```bash invoke test ``` -------------------------------- ### Start a UNIX Domain Socket Server Source: https://waterbutler.readthedocs.io/en/latest/_modules/asyncio/streams.html Initializes a server for UNIX domain sockets using a custom protocol factory. ```python async def start_unix_server(client_connected_cb, path=None, *, loop=None, limit=_DEFAULT_LIMIT, **kwds): """Similar to `start_server` but works with UNIX Domain Sockets.""" if loop is None: loop = events.get_event_loop() def factory(): reader = StreamReader(limit=limit, loop=loop) protocol = StreamReaderProtocol(reader, client_connected_cb, loop=loop) return protocol return await loop.create_unix_server(factory, path, **kwds) ``` -------------------------------- ### Define hierarchical configuration Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/settings.html Example of a hierarchical dictionary structure for server configuration. ```json "SERVER_CONFIG": { "HOST": "http://localhost:7777" } ``` -------------------------------- ### Get Info Source: https://waterbutler.readthedocs.io/en/latest/api.html Retrieves the contents of a folder or details of a particular file. ```APIDOC ## GET /resources//providers// ### Description Retrieves metadata for a file or folder. The response is a JSON-API formatted response. ### Method GET ### Endpoint /resources//providers// ### Parameters #### Query Parameters - **meta** (string) - Required - Used to trigger metadata retrieval. ``` -------------------------------- ### Get Info Request Source: https://waterbutler.readthedocs.io/en/latest/api.html Retrieve metadata for files or folders by appending the meta query parameter. ```text Method: GET Params: ?meta= Success: 200 OK + file representation Example: GET /resources/mst3k/providers/osfstorage/?meta= ``` -------------------------------- ### Run API tests with Newman Source: https://waterbutler.readthedocs.io/en/latest/testing.html Install Newman globally and execute a specific Postman collection with a provided environment file. ```bash npm install -g newman ``` ```bash newman run tests/postman/collections/copy_files.json -e copy_file_sample.json ``` -------------------------------- ### Define nested configuration structure Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/settings.html Example of a nested dictionary structure for analytics configuration. ```json "SERVER_CONFIG": { "ANALYTICS": { "PROJECT_ID": "foo" } } ``` -------------------------------- ### Download File Request Source: https://waterbutler.readthedocs.io/en/latest/api.html Download a file by issuing a GET request to its URL. ```text Method: GET Params: Success: 200 OK + file body Example: GET /resources/mst3k/providers/osfstorage/2348825492342 ``` -------------------------------- ### Get full environment variable key Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/settings.html Method to construct the environment variable name for a given key. ```python def full_key(self, key): """The name of the envvar which corresponds to this key.""" return '{}_{}'.format(self.parent, key) if self.parent else key ``` -------------------------------- ### Initialize SettingsDict Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/settings.html Constructor for the SettingsDict class. ```python def __init__(self, *args, parent=None, **kwargs): self.parent = parent super().__init__(*args, **kwargs) ``` -------------------------------- ### Load configuration file Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/settings.html Logic to determine the configuration path and load the settings from a JSON file. ```python try: config_path = os.environ['{}_CONFIG'.format(PROJECT_NAME.upper())] except KeyError: env = os.environ.get('ENV', 'test') config_path = '{}/{}-{}.json'.format(PROJECT_CONFIG_PATH, PROJECT_NAME, env) config = SettingsDict() config_path = os.path.expanduser(config_path) if not os.path.exists(config_path): logging.warning('No \'{}\' configuration file found'.format(config_path)) else: with open(os.path.expanduser(config_path)) as fp: config = SettingsDict(json.load(fp)) ``` -------------------------------- ### Initialize and use FormDataStream Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/core/streams/http.html Demonstrates how to instantiate FormDataStream and add fields or files to the multipart request body. ```python >>> stream = FormDataStream(key1='value1', file=FileStream(...)) ``` ```python >>> stream = FormDataStream() >>> stream.add_field('key1', 'value1') >>> stream.add_file('file', FileStream(...), mime='text/plain') ``` ```python >>> FormDataStream(fieldName=(FileStream(...), 'fileName', 'Mime', 'encoding')) ``` -------------------------------- ### Initialize FormDataStream Source: https://waterbutler.readthedocs.io/en/latest/waterbutler.core.streams.html Demonstrates basic initialization of the FormDataStream class. ```python >>> stream = FormDataStream(key1='value1', file=FileStream(...)) ``` -------------------------------- ### BaseProvider Method Requirements Source: https://waterbutler.readthedocs.io/en/latest/adding-providers.html List of methods to implement when creating a new provider, categorized by their implementation requirements. ```text validate_path() abstract validate_v1_path() abstract download() abstract metadata() abstract upload() abstract delete() abstract can_duplicate_names() abstract create_folder() error (405 Not Supported) intra_copy() error (501 Not Implemented) intra_move() default can_intra_copy() default can_intra_move() default exists() default revalidate_path() default zip() default path_from_metadata() default revisions() default shares_storage_root() default move() default copy() default handle_naming() default handle_name_conflict() default ``` -------------------------------- ### Open Network Connection Source: https://waterbutler.readthedocs.io/en/latest/_modules/asyncio/streams.html Establishes a connection and returns a (reader, writer) pair using StreamReader and StreamWriter. ```python async def open_connection(host=None, port=None, *, loop=None, limit=_DEFAULT_LIMIT, **kwds): """A wrapper for create_connection() returning a (reader, writer) pair. The reader returned is a StreamReader instance; the writer is a StreamWriter instance. The arguments are all the usual arguments to create_connection() except protocol_factory; most common are positional host and port, with various optional keyword arguments following. Additional optional keyword arguments are loop (to set the event loop instance to use) and limit (to set the buffer limit passed to the StreamReader). (If you want to customize the StreamReader and/or StreamReaderProtocol classes, just copy the code -- there's really nothing special here except some convenience.) """ if loop is None: loop = events.get_event_loop() reader = StreamReader(limit=limit, loop=loop) protocol = StreamReaderProtocol(reader, loop=loop) transport, _ = await loop.create_connection( lambda: protocol, host, port, **kwds) writer = StreamWriter(transport, protocol, reader, loop) return reader, writer ``` -------------------------------- ### Create hierarchical metrics records Source: https://waterbutler.readthedocs.io/en/latest/waterbutler.core.html Demonstrates how to create nested metrics records using MetricsRecord and MetricsSubRecord to track hierarchical data. ```python parent = MetricsRecord('foo') child = parent.new_subrecord('bar') grandchild = child.new_subrecord('baz') print(parent.key) # foo print(child.key) # foo_bar print(grandchild.key) # bar_baz ``` -------------------------------- ### Create Subfolder Request Source: https://waterbutler.readthedocs.io/en/latest/api.html Create a new subfolder using a PUT request with kind and name query parameters. ```text Method: PUT Query Params: ?kind=folder&name={new_folder_name} Body: Success: 201 Created + new folder representation Example: PUT /resources/mst3k/providers/osfstorage/?kind=folder&name=foo-folder ``` -------------------------------- ### Configure file options in FormDataStream Source: https://waterbutler.readthedocs.io/en/latest/waterbutler.core.streams.html Demonstrates passing additional file options as a tuple to the FormDataStream constructor. ```python >>> FormDataStream(fieldName=(FileStream(...), 'fileName', 'Mime', 'encoding')) ``` -------------------------------- ### Fetch configuration value Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/settings.html Method to retrieve a value, checking environment variables before the on-disk config. ```python def get(self, key, default=None): """Fetch a config value for ``key`` from the settings. First checks the env, then the on-disk config. If neither exists, returns ``default``.""" env = self.full_key(key) if env in os.environ: return os.environ.get(env) return super().get(key, default) ``` -------------------------------- ### Define default logging configuration Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/settings.html Configuration dictionary for logging, including conditional formatter selection. ```python try: import colorlog # noqa DEFAULT_FORMATTER = { '()': 'colorlog.ColoredFormatter', 'format': '%(cyan)s[%(asctime)s]%(log_color)s[%(levelname)s][%(name)s]: %(reset)s%(message)s' } except ImportError: DEFAULT_FORMATTER = { '()': 'waterbutler.core.logging.MaskFormatter', 'format': '[%(asctime)s][%(levelname)s][%(name)s]: %(message)s', 'pattern': '(?<=cookie=)(.*?)(?=&|$)', 'mask': '***' } DEFAULT_LOGGING_CONFIG = { 'version': 1, 'disable_existing_loggers': False, 'formatters': { 'console': DEFAULT_FORMATTER, }, 'handlers': { 'console': { 'class': 'logging.StreamHandler', 'level': 'INFO', 'formatter': 'console' }, 'syslog': { 'class': 'logging.handlers.SysLogHandler', 'level': 'INFO' } }, 'loggers': { '': { 'handlers': ['console'], 'level': 'INFO', 'propagate': False } }, 'root': { 'level': 'INFO', 'handlers': ['console'] } } ``` -------------------------------- ### Order Python imports by line length Source: https://waterbutler.readthedocs.io/en/latest/contributing.html Demonstrates the preferred import ordering style for Waterbutler, where imports are ordered by line length rather than alphabetically. ```python import abc import asyncio import itertools from urllib import parse import furl import aiohttp from waterbutler.core import streams from waterbutler.core import exceptions # Not import abc import asyncio import itertools from urllib import parse import aiohttp import furl from waterbutler.core import exceptions from waterbutler.core import streams ``` -------------------------------- ### Add fields and files to FormDataStream Source: https://waterbutler.readthedocs.io/en/latest/waterbutler.core.streams.html Demonstrates adding fields and files to an existing FormDataStream instance. ```python >>> stream = FormDataStream() >>> stream.add_field('key1', 'value1') >>> stream.add_file('file', FileStream(...), mime='text/plain') ``` -------------------------------- ### WaterButler Configuration Settings Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/settings.html Defines core application settings, logging configuration, and analytics integration parameters. ```python def child(key): return config.child(key) DEBUG = config.get_bool('DEBUG', True) OP_CONCURRENCY = int(config.get('OP_CONCURRENCY', 5)) logging_config = config.get('LOGGING', DEFAULT_LOGGING_CONFIG) logging.config.dictConfig(logging_config) SENTRY_DSN = config.get_nullable('SENTRY_DSN', None) analytics_config = config.child('ANALYTICS') MFR_IDENTIFYING_HEADER = analytics_config.get('MFR_IDENTIFYING_HEADER', 'X-Cos-Mfr-Render-Request') MFR_DOMAIN = analytics_config.get('MFR_DOMAIN', 'http://localhost:7778').rstrip('/') keen_config = analytics_config.child('KEEN') KEEN_API_BASE_URL = keen_config.get('API_BASE_URL', 'https://api.keen.io') KEEN_API_VERSION = keen_config.get('API_VERSION', '3.0') KEEN_ENABLE_LOGGING = keen_config.get_bool('ENABLE_LOGGING', True) keen_private_config = keen_config.child('PRIVATE') KEEN_PRIVATE_PROJECT_ID = keen_private_config.get_nullable('PROJECT_ID', None) KEEN_PRIVATE_WRITE_KEY = keen_private_config.get_nullable('WRITE_KEY', None) KEEN_PRIVATE_LOG_ACTIONS = keen_private_config.get_bool('LOG_ACTIONS', True) keen_public_config = keen_config.child('PUBLIC') KEEN_PUBLIC_PROJECT_ID = keen_public_config.get_nullable('PROJECT_ID', None) KEEN_PUBLIC_WRITE_KEY = keen_public_config.get_nullable('WRITE_KEY', None) KEEN_PUBLIC_LOG_ACTIONS = keen_public_config.get_bool('LOG_ACTIONS', True) WEBDAV_METHODS = {'PROPFIND', 'MKCOL', 'MOVE', 'COPY'} AIOHTTP_TIMEOUT = int(config.get('AIOHTTP_TIMEOUT', 3600)) # time in seconds ``` -------------------------------- ### Retrieve configuration values Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/settings.html Accessing nested configuration values using the SettingsDict class. ```python config = SettingsDict(json.load('local-config.json')) server_cfg = config.child('SERVER_CONFIG') host = server_cfg.get('HOST') ``` -------------------------------- ### Fetch nullable configuration value Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/settings.html Method to retrieve a value, treating empty strings as None. ```python def get_nullable(self, key, default=None): """Fetch a config value and interpret the empty string as None. Useful for external code that expects an explicit None.""" value = self.get(key, default) return None if value == '' else value ``` -------------------------------- ### Move and Copy File or Folder Source: https://waterbutler.readthedocs.io/en/latest/api.html Use a POST request to the move link to move or copy files and folders. The path parameter is required, while rename, conflict, resource, and provider are optional. ```text Method: POST Query Params: Body (JSON): { // mandatory "action": "move"|"copy", "path": {path_attribute_of_target_folder}, // optional "rename": {new_name}, "conflict": "replace"|"keep"|"warn", // defaults to 'warn' "resource": {node_id}, // defaults to current {node_id} "provider": {provider} // defaults to current {provider} } Success: 200 OK or 201 Created + new entity representation ``` -------------------------------- ### Fetch object configuration value Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/settings.html Method to retrieve a value and interpret it as a Python object or list. ```python def get_object(self, key, default=None): """Fetch a config value and interpret as a Python object or list. Since envvars are always strings, interpret values of type `str` as JSON object or array. Otherwise assume the type is already a python object.""" value = self.get(key, default) if isinstance(value, str): value = json.loads(value) return value ``` -------------------------------- ### Override configuration via environment variables Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/settings.html Overriding nested configuration values using environment variables with underscore-joined keys. ```bash $ SERVER_CONFIG_HOST='http://foo.bar.com' invoke server ``` -------------------------------- ### Constructing a v1 API URL Source: https://waterbutler.readthedocs.io/en/latest/api.html The base structure for v1 API requests, where node_id, provider_id, and id_or_path are required components. ```text http://files.osf.io/v1/resources//providers// e.g. http://files.osf.io/v1/resources/jy4bd/providers/osfstorage/523402a0234 ``` -------------------------------- ### Open UNIX Domain Socket Connection Source: https://waterbutler.readthedocs.io/en/latest/_modules/asyncio/streams.html Establishes a connection over a UNIX domain socket, similar to open_connection. ```python async def open_unix_connection(path=None, *, loop=None, limit=_DEFAULT_LIMIT, **kwds): """Similar to `open_connection` but works with UNIX Domain Sockets.""" if loop is None: loop = events.get_event_loop() reader = StreamReader(limit=limit, loop=loop) protocol = StreamReaderProtocol(reader, loop=loop) transport, _ = await loop.create_unix_connection( lambda: protocol, path, **kwds) writer = StreamWriter(transport, protocol, reader, loop) return reader, writer ``` -------------------------------- ### FileStreamReader and PartialFileStreamReader Implementation Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/core/streams/file.html Classes for reading files and specific byte ranges asynchronously. FileStreamReader handles full file reads, while PartialFileStreamReader restricts reads to a defined byte range. ```python import os import asyncio from waterbutler.core.streams.base import BaseStream [docs]class FileStreamReader(BaseStream): def __init__(self, file_pointer): super().__init__() self.file_gen = None self.file_pointer = file_pointer self.read_size = None self.content_type = 'application/octet-stream' @property def size(self): cursor = self.file_pointer.tell() self.file_pointer.seek(0, os.SEEK_END) ret = self.file_pointer.tell() self.file_pointer.seek(cursor) return ret [docs] def close(self): self.file_pointer.close() self.feed_eof() [docs] async def chunk_reader(self): self.file_pointer.seek(0) while True: chunk = self.file_pointer.read(self.read_size) if not chunk: self.feed_eof() yield b'' yield chunk async def _read(self, size): self.file_gen = self.file_gen or self.chunk_reader() self.read_size = size # add sleep of 0 so read will yield and continue in next io loop iteration # asyncio.sleep(0) yields None by default, which displeases tornado await asyncio.sleep(0.001) async for chunk in self.file_gen: return chunk class PartialFileStreamReader(FileStreamReader): """Awful class, used to avoid messing with FileStreamReader. Extends FSR with start and end byte offsets to indicate a byte range of the file to return. Reading from this stream will only return the requested range, never data outside of it. """ def __init__(self, file_pointer, byte_range): super().__init__(file_pointer) self.start = byte_range[0] self.end = byte_range[1] self.bytes_read = 0 @property def size(self): return self.end - self.start + 1 @property def total_size(self): cursor = self.file_pointer.tell() self.file_pointer.seek(0, os.SEEK_END) ret = self.file_pointer.tell() self.file_pointer.seek(cursor) return ret @property def partial(self): return self.size < self.total_size @property def content_range(self): return 'bytes {}-{}/{}'.format(self.start, self.end, self.total_size) async def chunk_reader(self): self.file_pointer.seek(self.start) while True: chunk = self.file_pointer.read(self.read_size) self.bytes_read += self.read_size if not chunk: self.feed_eof() yield b'' yield chunk async def _read(self, size): self.file_gen = self.file_gen or self.chunk_reader() bytes_remaining = self.size - self.bytes_read self.read_size = bytes_remaining if size == -1 else min(size, bytes_remaining) # add sleep of 0 so read will yield and continue in next io loop iteration # asyncio.sleep(0) yields None by default, which displeases tornado await asyncio.sleep(0.001) async for chunk in self.file_gen: return chunk ``` -------------------------------- ### Initialize BaseStream with cutoff Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/core/streams/base.html Constructor for a stream that limits reading to a specific cutoff size. ```python def __init__(self, stream, cutoff): super().__init__() self.stream = stream self._cutoff = cutoff self._thus_far = 0 self._size = min(cutoff, stream.size) ``` -------------------------------- ### Fetch sub-dictionary Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/settings.html Method to retrieve a sub-dictionary as a new SettingsDict instance. ```python def child(self, key): """Fetch a sub-dict of the current dict.""" return SettingsDict(self.get(key, {}), parent=self.full_key(key)) ``` -------------------------------- ### Download Source: https://waterbutler.readthedocs.io/en/latest/api.html Downloads a file or a zip archive of a folder. ```APIDOC ## GET /resources//providers// ### Description Downloads a file or a zip archive of a folder. The response includes a Content-Disposition header to trigger a browser download. ### Method GET ### Endpoint /resources//providers// ### Parameters #### Query Parameters - **zip** (string) - Optional - If provided, downloads a zip archive of the folder. ``` -------------------------------- ### Create Subfolder Source: https://waterbutler.readthedocs.io/en/latest/api.html Creates a new subfolder within an existing folder. ```APIDOC ## PUT /resources//providers// ### Description Creates a subfolder of an existing folder. Returns 201 Created on success or 409 Conflict if the name exists. ### Method PUT ### Parameters #### Query Parameters - **kind** (string) - Required - Must be 'folder'. - **name** (string) - Required - The name of the new folder. ``` -------------------------------- ### Upload New File Request Source: https://waterbutler.readthedocs.io/en/latest/api.html Upload a file to a folder using a PUT request with raw file data in the body. ```text Method: PUT Query Params: ?kind=file&name={new_file_name} Body (Raw): Success: 201 Created + new file representation Example: PUT /resources/mst3k/providers/osfstorage/?kind=file&name=foo-file ``` -------------------------------- ### Update Existing File Request Source: https://waterbutler.readthedocs.io/en/latest/api.html Update an existing file by sending a PUT request with raw file data and the kind query parameter. ```text Method: PUT Query Params: ?kind=file Body (Raw): Success: 200 OK + updated file representation Example: PUT /resources/mst3k/providers/osfstorage/2348825492342?kind=file ``` -------------------------------- ### Download Zip Archive Request Source: https://waterbutler.readthedocs.io/en/latest/api.html Download a folder as a zip archive by appending the zip query parameter. ```text Method: GET Params: Success: 200 OK + folder body Example: GET /resources/mst3k/providers/osfstorage/23488254123123/?zip= ``` -------------------------------- ### Data Signing Utility Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/core/signing.html Utility function to sign data with a time-to-live (TTL) constraint. ```python [docs]def sign_data(signer, data, ttl=100): target = {'time': int(time.time() + ttl)} target.update(data) payload, signature = signer.sign_payload(target) return { 'payload': payload.decode(), 'signature': signature, } ``` -------------------------------- ### Fetch boolean configuration value Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/settings.html Method to retrieve a value and interpret it as a boolean. ```python def get_bool(self, key, default=None): """Fetch a config value and interpret as a bool. Since envvars are always strings, interpret '0' and the empty string as False and '1' as True. Anything else is probably an acceident, so die screaming.""" value = self.get(key, default) if value in [False, 0, '0', '']: retval = False elif value in [True, 1, '1']: retval = True else: raise Exception( '{} should be a truthy value, but instead we got {}'.format( self.full_key(key), value ) ) return retval ``` -------------------------------- ### Debug ABC Registry and Caches Source: https://waterbutler.readthedocs.io/en/latest/_modules/abc.html Helper functions for inspecting and clearing the internal registry and cache states of an ABC. ```python def _dump_registry(cls, file=None): """Debug helper to print the ABC registry.""" print(f"Class: {cls.__module__}.{cls.__qualname__}", file=file) print(f"Inv. counter: {get_cache_token()}", file=file) (_abc_registry, _abc_cache, _abc_negative_cache, _abc_negative_cache_version) = _get_dump(cls) print(f"_abc_registry: {_abc_registry!r}", file=file) print(f"_abc_cache: {_abc_cache!r}", file=file) print(f"_abc_negative_cache: {_abc_negative_cache!r}", file=file) print(f"_abc_negative_cache_version: {_abc_negative_cache_version!r}", file=file) def _abc_registry_clear(cls): """Clear the registry (for debugging or testing).""" _reset_registry(cls) def _abc_caches_clear(cls): """Clear the caches (for debugging or testing).""" _reset_caches(cls) ``` -------------------------------- ### Upload New File Source: https://waterbutler.readthedocs.io/en/latest/api.html Uploads a new file to a folder. ```APIDOC ## PUT /resources//providers// ### Description Uploads a new file to a folder using raw file data in the request body. ### Method PUT ### Parameters #### Query Parameters - **kind** (string) - Required - Must be 'file'. - **name** (string) - Required - The name of the new file. ### Request Body - **raw data** (binary) - Required - The file content. ``` -------------------------------- ### Implement readexactly and asynchronous iteration Source: https://waterbutler.readthedocs.io/en/latest/_modules/asyncio/streams.html Provides the logic for reading a specific number of bytes from a stream and enables asynchronous iteration over stream lines. ```python If stream was paused, this function will automatically resume it if needed. """ if n < 0: raise ValueError('readexactly size can not be less than zero') if self._exception is not None: raise self._exception if n == 0: return b'' while len(self._buffer) < n: if self._eof: incomplete = bytes(self._buffer) self._buffer.clear() raise IncompleteReadError(incomplete, n) await self._wait_for_data('readexactly') if len(self._buffer) == n: data = bytes(self._buffer) self._buffer.clear() else: data = bytes(self._buffer[:n]) del self._buffer[:n] self._maybe_resume_transport() return data def __aiter__(self): return self async def __anext__(self): val = await self.readline() if val == b'': raise StopAsyncIteration return val ``` -------------------------------- ### Signer Class Implementation Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/core/signing.html A class for signing and verifying messages and payloads using HMAC. ```python [docs]class Signer(object): def __init__(self, secret, digest): assert callable(digest) self.secret = secret self.digest = digest [docs] def sign_message(self, message): return hmac.new( key=self.secret, digestmod=self.digest, msg=message, ).hexdigest() [docs] def sign_payload(self, payload): message = serialize_payload(payload) signature = self.sign_message(message) return message, signature [docs] def verify_message(self, signature, message): expected = self.sign_message(message) return signature == expected [docs] def verify_payload(self, signature, payload): _, expected = self.sign_payload(payload) return signature == expected ``` -------------------------------- ### Update Existing File Source: https://waterbutler.readthedocs.io/en/latest/api.html Updates an existing file by creating a new version. ```APIDOC ## PUT /resources//providers// ### Description Updates an existing file with new content. Returns 200 OK. ### Method PUT ### Parameters #### Query Parameters - **kind** (string) - Required - Must be 'file'. ### Request Body - **raw data** (binary) - Required - The new file content. ``` -------------------------------- ### POST Move & Copy Source: https://waterbutler.readthedocs.io/en/latest/api.html Moves or copies a file or folder to a specified destination. Supports renaming and conflict resolution strategies. ```APIDOC ## POST /move ### Description Moves or copies a file or folder to a target location. Can also be used to rename items during the transfer. ### Method POST ### Request Body - **action** (string) - Required - The operation to perform: "move" or "copy". - **path** (string) - Required - The path attribute of the target folder. - **rename** (string) - Optional - The new name for the file or folder. - **conflict** (string) - Optional - Conflict resolution strategy: "replace", "keep", or "warn". Defaults to "warn". - **resource** (string) - Optional - The node ID of the target. Defaults to current node. - **provider** (string) - Optional - The storage provider. Defaults to current provider. ### Response #### Success Response (200/201) - **entity** (object) - Metadata of the moved or copied file or folder. ``` -------------------------------- ### StreamReaderProtocol Implementation Source: https://waterbutler.readthedocs.io/en/latest/_modules/asyncio/streams.html Adapts between standard asyncio protocols and StreamReader objects. ```python class StreamReaderProtocol(FlowControlMixin, protocols.Protocol): """Helper class to adapt between Protocol and StreamReader. (This is a helper class instead of making StreamReader itself a Protocol subclass, because the StreamReader has other potential uses, and to prevent the user of the StreamReader to accidentally call inappropriate methods of the protocol.) """ def __init__(self, stream_reader, client_connected_cb=None, loop=None): super().__init__(loop=loop) self._stream_reader = stream_reader self._stream_writer = None self._client_connected_cb = client_connected_cb self._over_ssl = False self._closed = self._loop.create_future() def connection_made(self, transport): self._stream_reader.set_transport(transport) self._over_ssl = transport.get_extra_info('sslcontext') is not None if self._client_connected_cb is not None: self._stream_writer = StreamWriter(transport, self, self._stream_reader, self._loop) res = self._client_connected_cb(self._stream_reader, self._stream_writer) if coroutines.iscoroutine(res): self._loop.create_task(res) def connection_lost(self, exc): if self._stream_reader is not None: if exc is None: self._stream_reader.feed_eof() else: self._stream_reader.set_exception(exc) if not self._closed.done(): if exc is None: self._closed.set_result(None) else: self._closed.set_exception(exc) super().connection_lost(exc) self._stream_reader = None self._stream_writer = None def data_received(self, data): self._stream_reader.feed_data(data) def eof_received(self): self._stream_reader.feed_eof() if self._over_ssl: # Prevent a warning in SSLProtocol.eof_received: # "returning true from eof_received() # has no effect when using ssl" return False return True ``` -------------------------------- ### CORS Handling Mixin Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/server/utils.html Provides methods to set default CORS headers and handle OPTIONS requests in Tornado handlers. ```python class CORsMixin: def _cross_origin_is_allowed(self): if self.request.method == 'OPTIONS': return True elif not self.request.cookies and self.request.headers.get('Authorization'): return True return False def set_default_headers(self): if not self.request.headers.get('Origin'): return allowed_origin = None if self._cross_origin_is_allowed(): allowed_origin = self.request.headers['Origin'] elif isinstance(settings.CORS_ALLOW_ORIGIN, str): if settings.CORS_ALLOW_ORIGIN == '*': # Wild cards cannot be used with allowCredentials. # Match Origin if its specified, makes pdfs and pdbs render properly allowed_origin = self.request.headers['Origin'] else: allowed_origin = settings.CORS_ALLOW_ORIGIN else: if self.request.headers['Origin'] in settings.CORS_ALLOW_ORIGIN: allowed_origin = self.request.headers['Origin'] if allowed_origin is not None: self.set_header('Access-Control-Allow-Origin', allowed_origin) self.set_header('Access-Control-Allow-Credentials', 'true') self.set_header('Access-Control-Allow-Headers', ', '.join(CORS_ACCEPT_HEADERS)) self.set_header('Access-Control-Expose-Headers', ', '.join(CORS_EXPOSE_HEADERS)) self.set_header('Cache-control', 'no-store, no-cache, must-revalidate, max-age=0') def options(self, *args, **kwargs): self.set_status(204) if self.request.headers.get('Origin'): self.set_header('Access-Control-Allow-Methods', 'GET, PUT, POST, DELETE'), ``` -------------------------------- ### Read until separator implementation Source: https://waterbutler.readthedocs.io/en/latest/_modules/asyncio/streams.html Internal logic for reading from a stream until a specific separator is encountered, including buffer limit checks. ```python # `offset` is the number of bytes from the beginning of the buffer # where there is no occurrence of `separator`. offset = 0 # Loop until we find `separator` in the buffer, exceed the buffer size, # or an EOF has happened. while True: buflen = len(self._buffer) # Check if we now have enough data in the buffer for `separator` to # fit. if buflen - offset >= seplen: isep = self._buffer.find(separator, offset) if isep != -1: # `separator` is in the buffer. `isep` will be used later # to retrieve the data. break # see upper comment for explanation. offset = buflen + 1 - seplen if offset > self._limit: raise LimitOverrunError( 'Separator is not found, and chunk exceed the limit', offset) # Complete message (with full separator) may be present in buffer # even when EOF flag is set. This may happen when the last chunk # adds data which makes separator be found. That's why we check for # EOF *ater* inspecting the buffer. if self._eof: chunk = bytes(self._buffer) self._buffer.clear() raise IncompleteReadError(chunk, None) # _wait_for_data() will resume reading if stream was paused. await self._wait_for_data('readuntil') if isep > self._limit: raise LimitOverrunError( 'Separator is found, but chunk is longer than limit', isep) chunk = self._buffer[:isep + seplen] del self._buffer[:isep + seplen] self._maybe_resume_transport() return bytes(chunk) ``` -------------------------------- ### Utility Mixin for Status Codes Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/server/utils.html Extends Tornado's set_status to support custom HTTP reason phrases defined in the module. ```python class UtilMixin: bytes_downloaded = 0 bytes_uploaded = 0 def set_status(self, code, reason=None): return super().set_status(code, reason or HTTP_REASONS.get(code)) ``` -------------------------------- ### Delete File or Folder Source: https://waterbutler.readthedocs.io/en/latest/api.html Send a DELETE request to the delete link to remove a file or folder. The confirm_delete query parameter is required when deleting the root folder. ```text Method: DELETE Query Params: ?confirm_delete=1 // required for root folder delete only Success: 204 No Content ``` -------------------------------- ### MetricsBase Class Definition Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/core/metrics.html A wrapper class for managing metrics stored in a dictionary, supporting dot-notation for nested keys. ```python [docs]class MetricsBase(): """Lightweight wrapper around a dict to make keeping track of metrics a little easier. Current functionality is limited, but may be extended later. To do: * update/override method to indicate expectations of existing key self.metrics.add_default('some.flag', True) self.metrics.override('some.flag', False) # dies if 'some.flag' doesn't already exist * optional type validation? self.metrics.add('some.flag', True, bool()) -or- self.metrics.define('some.flag', bool()) self.metrics.add('some.flag', 'foobar') # dies, 'foobar' isn't a bool """ def __init__(self): self._metrics = {} [docs] def key(self): """ID string for this object""" raise NotImplementedError [docs] def add(self, key, value): """add() stores the given value under the given key. Subkeys can be specified by placing a dot between the parent and child keys. e.g. 'foo.bar' will be interpreted as ``self._metrics['foo']['bar']`` :param str key: the key to store ``value`` under :param value: the value to store, type unrestricted """ self._set_dotted_key(self._metrics, key, value) [docs] def incr(self, key): """incr() increments the value stored in key, or initializes it to 1 if it has not yet been set. :param str key: the key to increment the ``value`` of """ value = self._get_dotted_key(self._metrics, key) self._set_dotted_key(self._metrics, key, 1 if value is None else value + 1) [docs] def append(self, key, new_value): """Assume key points to a list and append ``new_value`` to it. Will initialize a list if key is undefined. Type homogeneity of list members is not enforced. :param str key: the key to store ``value`` under :param value: the value to store, type unrestricted """ old_value = self._get_dotted_key(self._metrics, key) self._set_dotted_key(self._metrics, key, ([] if old_value is None else old_value) + [new_value]) [docs] def merge(self, record): """Merges a dict into the current metrics. :param dict record: a dict to merge with the current metrics """ _merge_dicts(self._metrics, record) [docs] def serialize(self): """Return a copy of the metrics""" return copy.deepcopy(self._metrics) [docs] def manifesto(self): """'This is who I am and this is what I stand for!' Returns a dict with one entry: our key pointing to our metrics """ return {self.key: self.serialize()} def _get_dotted_key(self, store, key): """Naive method to get a nested dict values via dot-separated keys. e.g ``_get_dotted_keys(self._metrics, 'foo.bar')`` is equivalent to ``return self._metrics['foo']['bar']``, but will autovivify any non-existing intermediate dicts and will return ``None`` if the final part does not exist. This method is neither resilient nor intelligent and will react with bad grace if one of the intermediate keys already exists and is not a dict key. """ parts = key.split('.') current = store for part in parts[:-1]: if part not in current: current[part] = {} current = current[part] return current.get(parts[-1], None) def _set_dotted_key(self, store, key, value): """Naive method to set nested dict values via dot-separated keys. e.g ``_set_dotted_keys(self._metrics, 'foo.bar', 'moo')`` is equivalent to ``self._metrics['foo']['bar'] = 'moo'``. This method is neither resilient nor intelligent and will react with bad grace if one of the keys already exists and is not a dict key. """ parts = key.split('.') current = store for part in parts[:-1]: if part not in current: current[part] = {} current = current[part] current[parts[-1]] = value ``` -------------------------------- ### Write stream utility Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/server/utils.html Handles asynchronous stream writing with chunking and flush operations. Includes a temporary fix for bytearray compatibility. ```python async def write_stream(self, stream): try: while True: chunk = await stream.read(settings.CHUNK_SIZE) if not chunk: break # Temp fix, write does not accept bytearrays currently if isinstance(chunk, bytearray): chunk = bytes(chunk) self.write(chunk) self.bytes_downloaded += len(chunk) del chunk await self.flush() except tornado.iostream.StreamClosedError: # Client has disconnected early. # No need for any exception to be raised return ``` -------------------------------- ### Rename Source: https://waterbutler.readthedocs.io/en/latest/api.html Renames a file or folder. ```APIDOC ## POST /resources//providers// ### Description Renames an existing file or folder. ### Method POST ### Request Body - **action** (string) - Required - Must be 'rename'. - **rename** (string) - Required - The new name for the entity. ``` -------------------------------- ### Magic Query Parameters Source: https://waterbutler.readthedocs.io/en/latest/api.html A collection of query parameters that can be appended to API requests to modify provider behavior. ```APIDOC ## Magic Query Parameters ### Description These query parameters apply to all providers and are used to specify operations such as metadata retrieval, downloading, uploading, and deletion. ### Parameters #### Query Parameters - **meta** (flag) - Optional - Returns file metadata instead of downloading contents. - **zip** (flag) - Optional - Downloads folder contents as a .zip file. - **kind** (string) - Optional - Specifies 'file' or 'folder' for PUT requests. - **name** (string) - Optional - Specifies the name for new files or folders in PUT requests. - **revisions / versions** (flag) - Optional - Returns metadata for all available file revisions. - **revision / version** (int) - Optional - Specifies the ID of a specific file or folder version. - **direct** (flag) - Optional - Forces WaterButler to act as a middleman for downloads. - **displayName** (string) - Optional - Sets the filename in the Content-Disposition header. - **mode** (string) - Optional - Indicates if a file is being downloaded for rendering. - **confirm_delete** (bool) - Optional - Required for recursive deletion of root folder contents. ``` -------------------------------- ### Implement MaskFormatter for log masking Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/core/logging.html A custom logging formatter that masks sensitive data based on a provided regex pattern. ```python import re import logging [docs]class MaskFormatter(logging.Formatter): def __init__(self, fmt=None, datefmt=None, style='%', pattern=None, mask='***'): logging.Formatter.__init__(self, fmt=fmt, datefmt=datefmt, style=style) self.pattern = re.compile(pattern) self.mask = mask [docs] def format(self, record): result = super().format(record) return self.pattern.sub(self.mask, result) ``` -------------------------------- ### Define an abstract method Source: https://waterbutler.readthedocs.io/en/latest/_modules/abc.html Use the abstractmethod decorator to mark a method as abstract. Classes using this must have a metaclass derived from ABCMeta. ```python class C(metaclass=ABCMeta): @abstractmethod def my_abstract_method(self, ...): ... ``` -------------------------------- ### Implement asynchronous iteration for BaseStream Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/core/streams/base.html Asynchronous iteration support for stream objects. ```python def __aiter__(self): return self # TODO: Add more note on `AsyncIterablePayload` and its `write()` method in aiohttp3 # TODO: Improve the BaseStream with `aiohttp.streams.AsyncStreamReaderMixin` async def __anext__(self): try: chunk = await self.read(CHUNK_SIZE) except EOFError: raise StopAsyncIteration if chunk == b'': raise StopAsyncIteration return chunk ``` -------------------------------- ### DELETE File or Folder Source: https://waterbutler.readthedocs.io/en/latest/api.html Deletes a file or folder. Root folder deletion requires a confirmation parameter. ```APIDOC ## DELETE /delete ### Description Deletes a file or folder. Deleting a root folder requires the confirm_delete query parameter. ### Method DELETE ### Query Parameters - **confirm_delete** (integer) - Required for root folder - Must be set to 1 to confirm deletion of root folder contents. ### Response #### Success Response (204) - No content returned. ``` -------------------------------- ### waterbutler.core.signing.sign_data Source: https://waterbutler.readthedocs.io/en/latest/waterbutler.core.html Signs data using a provided signer instance with a specified time-to-live. ```APIDOC ## sign_data(signer, data, ttl=100) ### Description Signs the provided data using the given signer instance, with a default time-to-live (ttl) of 100. ``` -------------------------------- ### Define Authentication Types and Base Handler Source: https://waterbutler.readthedocs.io/en/latest/_modules/waterbutler/core/auth.html Defines the AuthType enum for source/destination roles and the BaseAuthHandler abstract class for implementing authentication logic. ```python import abc from enum import Enum [docs]class AuthType(Enum): SOURCE = 0 DESTINATION = 1 [docs]class BaseAuthHandler(metaclass=abc.ABCMeta): [docs] @abc.abstractmethod async def fetch(self, request, bundle): pass [docs] @abc.abstractmethod async def get(self, resource, provider, request, action=None, auth_type=AuthType.SOURCE, path='', version=None): pass ``` -------------------------------- ### waterbutler.core.signing.order_recursive Source: https://waterbutler.readthedocs.io/en/latest/waterbutler.core.html Recursively sorts keys of input data and nested dictionaries to ensure consistent JSON payload ordering. ```APIDOC ## order_recursive(data) ### Description Recursively sort keys of input data and all its nested dictionaries. Used to ensure consistent ordering of JSON payloads. ```