<commit_before>import six
from six.moves import queue as Queue
from collections import defaultdict
import json
import os.path
import posixpath
import re
from functools import partial

from tqdm import tqdm

from cloudvolume import compression
from cloudvolume.exceptions import UnsupportedProtocolError
from cloudvolume.lib import mkdir, scatter
from cloudvolume.threaded_queue import ThreadedQueue, DEFAULT_THREADS
from cloudvolume.scheduler import schedule_green_jobs

import cloudvolume.paths

from .storage_interfaces import (
  FileInterface, HttpInterface, 
  S3Interface, GoogleCloudStorageInterface
)

def get_interface_class(protocol):
  if protocol == 'file':
    return FileInterface
  elif protocol == 'gs':
    return GoogleCloudStorageInterface
  elif protocol in ('s3', 'matrix'):
    return S3Interface
  elif protocol in ('http', 'https'):
    return HttpInterface
  else:
    raise UnsupportedProtocolError(str(self._path))

class StorageBase(object):
  """Abastract base class of Storage with some implementation details."""
  def __init__(self, layer_path, progress=False):
    self.progress = progress

    self._layer_path = layer_path
    self._path = cloudvolume.paths.extract(layer_path)
    self._interface_cls = get_interface_class(self._path.protocol)
  
  @property
  def layer_path(self):
    return self._layer_path

  def get_connection(self):
    return self._interface_cls(self._path)

  def get_path_to_file(self, file_path):
    return posixpath.join(self._layer_path, file_path)

  def put_json(self, file_path, content, content_type='application/json', *args, **kwargs):
    if type(content) != str:
      content = json.dumps(content)
    return self.put_file(file_path, content, content_type=content_type, *args, **kwargs)
    
  def get_json(self, file_path):
    content = self.get_file(file_path)
    if content is None:
      return None
    return json.loads(content.decode('utf8'))

  def put_file(self, file_path, content, content_type=None, compress=None, cache_control=None):
    """ 
    Args:
      filename (string): it can contains folders
      content (string): binary data to save
    """
    return self.put_files([ (file_path, content) ], 
      content_type=content_type, 
      compress=compress, 
      cache_control=cache_control, 
      block=False
    )

  def exists(self, file_path):
    raise NotImplementedError()

  def files_exist(self, file_paths):
    raise NotImplementedError()

  def get_file(self, file_path):
    raise NotImplementedError()

  def get_files(self, file_paths):
    raise NotImplementedError()

  def delete_file(self, file_path):
    raise NotImplementedError()

  def delete_files(self, file_paths):
    raise NotImplementedError()

  def list_files(self, prefix="", flat=False):
    raise NotImplementedError()

  def __del__(self):
    pass

  def __enter__(self):
    return self

  def __exit__(self, exception_type, exception_value, traceback):
    pass

class SimpleStorage(StorageBase):
  """
  Access files stored in Google Storage (gs), Amazon S3 (s3), 
  or the local Filesystem (file).

  e.g. with Storage('gs://bucket/dataset/layer') as stor:
      files = stor.get_file('filename')

  Required:
    layer_path (str): A protocol prefixed path of the above format.
      Accepts s3:// gs:// and file://. File paths are absolute.

  Optional:
    n_threads (int:20): number of threads to use downloading and uplaoding.
      If 0, execution will be on the main python thread.
    progress (bool:false): Show a tqdm progress bar for multiple 
      uploads and downloads.
  """
  def __init__(self, layer_path, progress=False):
    super(SimpleStorage, self).__init__(layer_path, progress)
    self._interface = self.get_connection()

  def put_files(self, files, content_type=None, compress=None, cache_control=None, block=True):
    """
    Put lots of files at once and get a nice progress bar. It'll also wait
    for the upload to complete, just like get_files.

    Required:
      files: [ (filepath, content), .... ]
    """
    for path, content in tqdm(files, disable=(not self.progress), desc="Uploading"):<commit_msg>docs: remove non-existent parameter from SimpleStorage docstring<commit_after>import six
from six.moves import queue as Queue
from collections import defaultdict
import json
import os.path
import posixpath
import re
from functools import partial

from tqdm import tqdm

from cloudvolume import compression
from cloudvolume.exceptions import UnsupportedProtocolError
from cloudvolume.lib import mkdir, scatter
from cloudvolume.threaded_queue import ThreadedQueue, DEFAULT_THREADS
from cloudvolume.scheduler import schedule_green_jobs

import cloudvolume.paths

from .storage_interfaces import (
  FileInterface, HttpInterface, 
  S3Interface, GoogleCloudStorageInterface
)

def get_interface_class(protocol):
  if protocol == 'file':
    return FileInterface
  elif protocol == 'gs':
    return GoogleCloudStorageInterface
  elif protocol in ('s3', 'matrix'):
    return S3Interface
  elif protocol in ('http', 'https'):
    return HttpInterface
  else:
    raise UnsupportedProtocolError(str(self._path))

class StorageBase(object):
  """Abastract base class of Storage with some implementation details."""
  def __init__(self, layer_path, progress=False):
    self.progress = progress

    self._layer_path = layer_path
    self._path = cloudvolume.paths.extract(layer_path)
    self._interface_cls = get_interface_class(self._path.protocol)
  
  @property
  def layer_path(self):
    return self._layer_path

  def get_connection(self):
    return self._interface_cls(self._path)

  def get_path_to_file(self, file_path):
    return posixpath.join(self._layer_path, file_path)

  def put_json(self, file_path, content, content_type='application/json', *args, **kwargs):
    if type(content) != str:
      content = json.dumps(content)
    return self.put_file(file_path, content, content_type=content_type, *args, **kwargs)
    
  def get_json(self, file_path):
    content = self.get_file(file_path)
    if content is None:
      return None
    return json.loads(content.decode('utf8'))

  def put_file(self, file_path, content, content_type=None, compress=None, cache_control=None):
    """ 
    Args:
      filename (string): it can contains folders
      content (string): binary data to save
    """
    return self.put_files([ (file_path, content) ], 
      content_type=content_type, 
      compress=compress, 
      cache_control=cache_control, 
      block=False
    )

  def exists(self, file_path):
    raise NotImplementedError()

  def files_exist(self, file_paths):
    raise NotImplementedError()

  def get_file(self, file_path):
    raise NotImplementedError()

  def get_files(self, file_paths):
    raise NotImplementedError()

  def delete_file(self, file_path):
    raise NotImplementedError()

  def delete_files(self, file_paths):
    raise NotImplementedError()

  def list_files(self, prefix="", flat=False):
    raise NotImplementedError()

  def __del__(self):
    pass

  def __enter__(self):
    return self

  def __exit__(self, exception_type, exception_value, traceback):
    pass

class SimpleStorage(StorageBase):
  """
  Access files stored in Google Storage (gs), Amazon S3 (s3), 
  or the local Filesystem (file).

  e.g. with Storage('gs://bucket/dataset/layer') as stor:
      files = stor.get_file('filename')

  Required:
    layer_path (str): A protocol prefixed path of the above format.
      Accepts s3:// gs:// and file://. File paths are absolute.

  Optional:
    progress (bool:false): Show a tqdm progress bar for multiple 
      uploads and downloads.
  """
  def __init__(self, layer_path, progress=False):
    super(SimpleStorage, self).__init__(layer_path, progress)
    self._interface = self.get_connection()

  def put_files(self, files, content_type=None, compress=None, cache_control=None, block=True):
    """
    Put lots of files at once and get a nice progress bar. It'll also wait
    for the upload to complete, just like get_files.

    Required:
      files: [ (filepath, content), .... ]
    """
    for path, content in tqdm(files, disable=(not self.progress), desc="Uploading"):
{'fraction_non_alphanumeric': 0.09396383186705767, 'fraction_numerical': 0.0020772238514173997, 'mean_word_length': 4.141331658291457, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7195852', 'n_tokens_mistral': 2446, 'n_tokens_neox': 2376, 'n_words': 709}