Differential D3334 Diff 11842 swh/storage/__init__.py

Changeset View

Standalone View

swh/storage/init.py

# Copyright (C) 2015-2020 The Software Heritage developers		# Copyright (C) 2015-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution		# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version		# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information		# See top-level LICENSE file for more information

import warnings		import warnings


STORAGE_IMPLEMENTATION = {		STORAGE_IMPLEMENTATION = {
"pipeline",		"pipeline",
"local",		"local",
"remote",		"remote",
"memory",		"memory",
"filter",		"filter",
"buffer",		"buffer",
"retry",		"retry",
		"tenacious",
"validate",		"validate",
"cassandra",		"cassandra",
}		}


def get_storage(cls, **kwargs):		def get_storage(cls, **kwargs):
"""Get a storage object of class `storage_class` with arguments		"""Get a storage object of class `storage_class` with arguments
`storage_args`.		`storage_args`.
Show All 36 Lines	def get_storage(cls, **kwargs):
elif cls == "memory":		elif cls == "memory":
from .in_memory import InMemoryStorage as Storage		from .in_memory import InMemoryStorage as Storage
elif cls == "filter":		elif cls == "filter":
from .filter import FilteringProxyStorage as Storage		from .filter import FilteringProxyStorage as Storage
elif cls == "buffer":		elif cls == "buffer":
from .buffer import BufferingProxyStorage as Storage		from .buffer import BufferingProxyStorage as Storage
elif cls == "retry":		elif cls == "retry":
from .retry import RetryingProxyStorage as Storage		from .retry import RetryingProxyStorage as Storage
		elif cls == "tenacious":
		from .tenacious import TenaciousProxyStorage as Storage
elif cls == "validate":		elif cls == "validate":
from .validate import ValidatingProxyStorage as Storage		from .validate import ValidatingProxyStorage as Storage

return Storage(**kwargs)		return Storage(**kwargs)


def get_storage_pipeline(steps):		def get_storage_pipeline(steps):
"""Recursively get a storage object that may use other storage objects		"""Recursively get a storage object that may use other storage objects
Show All 28 Lines