How to use the ipfshttpclient.multipart.DirectoryStream function in ipfshttpclient

To help you get started, we’ve selected a few ipfshttpclient examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github ipfs / py-ipfs-http-client / test / unit / test_multipart.py View on Github external
def test_body(self):
		"""Check the multipart HTTP body for the streamed directory."""
		# Get OS-agnostic path to test files
		path = os.path.join(os.path.dirname(os.path.dirname(__file__)),
		                    "functional", "fake_dir")
		instance = ipfshttpclient.multipart.DirectoryStream(path)
		expected = b"^(--\S+\r\nContent-Disposition: file; filename=\"\S+\""\
			+ b"\r\nContent-Type: application/\S+\r\n\r\n(.|\n)*"\
			+ b"\r\n)+--\S+--\r\n$"
		actual = b"".join(instance.body())
		assert re.search(expected, actual)
github ipfs / py-ipfs-http-client / ipfshttpclient / multipart.py View on Github external
def stream_directory_impl(directory, dirname=None):
		stream = DirectoryStream(directory,
		                         recursive=recursive, patterns=patterns,
		                         dirname=dirname, chunk_size=chunk_size)
		return stream.body(), stream.headers()
github ipfs / py-ipfs-http-client / ipfshttpclient / multipart.py View on Github external
def __init__(self, directory,
	             recursive=False, patterns='**', dirname=None,
	             chunk_size=default_chunk_size):
		self.patterns = []
		patterns = [patterns] if isinstance(patterns, str) else patterns
		for pattern in patterns:
			self.patterns.append(glob_compile(pattern) if isinstance(pattern, str) else pattern)

		self.directory = utils.convert_path(directory)
		if not isinstance(self.directory, int):
			self.directory = os.path.normpath(self.directory)
		self.recursive = recursive
		self.dirname   = dirname

		name = os.path.basename(self.directory) if not isinstance(self.directory, int) else ""
		super(DirectoryStream, self).__init__(name, chunk_size=chunk_size)