File Manager Factory

Use the factory for creating a file manager.

This is the main entry point for using the numerous files module. Use this method to get a file manager instance based on the value of the NUMEROUS_FILES_BACKEND environment variable.

The file manager instance can be used to perform file management operations such as reading, writing, and deleting files.

Parameters:
  • file_manager –

    File manager type.

  • **kwargs (dict[str, str], default: {} ) –

    Keyword arguments to pass to the file
Source code in numerous/files/__init__.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def file_manager_factory(**kwargs:dict[str,str]) -> AwsS3|Memory|LocalFolder:
    """
    Use the factory for creating a file manager.

    This is the main entry point for using the numerous files module.
    Use this method to get a file manager instance based on the value
    of the `NUMEROUS_FILES_BACKEND` environment variable.

    The file manager instance can be used to perform file management
    operations such as reading, writing, and deleting files.

    Args:
        file_manager: File manager type.
        **kwargs: Keyword arguments to pass to the file

    """
    # Get the file manager type from the environment.
    file_manager = os.environ.get("NUMEROUS_FILES_BACKEND", "INMEMORY")


    if file_manager == "INMEMORY":
        return Memory(**kwargs)

    if file_manager == "AWS_S3":
        env_var_aws_access_key = str(kwargs.get("aws_access_key_id",
                                                "NUMEROUS_FILES_AWS_ACCESS_KEY"))
        env_var_aws_secret_key = str(
            kwargs.get("aws_secret_access_key", "NUMEROUS_FILES_AWS_SECRET_KEY"))
        env_var_bucket = str(
            kwargs.get("bucket", "NUMEROUS_FILES_BUCKET"))
        env_var_base_prefix = str(kwargs.get(
            "base_prefix", "NUMEROUS_FILES_BASE_PREFIX"))

        # Get the credentials from the environment.
        if env_var_aws_access_key in os.environ \
                and env_var_aws_secret_key in os.environ:
            credentials = {
                "aws_access_key_id": os.environ[env_var_aws_access_key],
                "aws_secret_access_key": os.environ[env_var_aws_secret_key],
            }
        else:
            credentials = None
        # If the credentials are not in the environment,
        # set them to None so the aws client will use default auth.

        # Get the bucket from the environment.
        bucket = os.getenv(env_var_bucket, "numerous-files")
        # Get the base prefix from the environment.
        base_prefix = os.environ.get(env_var_base_prefix, "files")

        env_params = {
            "bucket": bucket,
            "base_prefix": base_prefix,
        }
        env_params.update(kwargs) # type: ignore[arg-type]

        return AwsS3(bucket=env_params["bucket"],
                     base_prefix=env_params["base_prefix"], credentials=credentials)

    if file_manager == "LOCAL":
        # Get a temporary folder from the operating system.

        return LocalFolder(workfolder=str(kwargs.get("workfolder", "./tmp")))

    err_str = f"Unknown file manager type {file_manager}"
    raise ValueError(err_str)