We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Support for general web URI address for hosted KFServing model location
/kind feature

**Describe the solution you'd like**

Currently there are 4 types of storage URIs that are accepted in KFServing yaml files:

https://github.com/kubeflow/kfserving/blob/e25758f6da020e3e8ad63f0880ab95de6af8395b/python/kfserving/kfserving/storage.py#L26-L30

(1) GCS
(2) AWS S3
(3) Azure (blob over https://)
(4) local (file://)

I think it would be useful to be able reference a model file hosted anywhere (via general web URI, aka "http://", such as a file on github:

https://github.com/jpatanooga/kubeflow_ops_book_dev/blob/master/kfserving/models/sklearn/iris_sklearn_svm_model.joblib

(in this example the raw file would be downloaded from the url: https://github.com/jpatanooga/kubeflow_ops_book_dev/blob/master/kfserving/models/sklearn/iris_sklearn_svm_model.joblib?raw=true )

This would allow for examples to be executed more efficiently; the KFServing project could even provide a repo of the pre-built models (perhaps), so that example model deployments could be easier to run for new users.




--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `python/kfserving/kfserving/storage.py`
Content:
```
1 # Copyright 2020 kubeflow.org.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #    http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 import glob
16 import logging
17 import tempfile
18 import os
19 import re
20 from urllib.parse import urlparse
21 from azure.storage.blob import BlockBlobService
22 from google.auth import exceptions
23 from google.cloud import storage
24 from minio import Minio
25 
26 _GCS_PREFIX = "gs://"
27 _S3_PREFIX = "s3://"
28 _BLOB_RE = "https://(.+?).blob.core.windows.net/(.+)"
29 _LOCAL_PREFIX = "file://"
30 
31 
32 class Storage(object): # pylint: disable=too-few-public-methods
33     @staticmethod
34     def download(uri: str, out_dir: str = None) -> str:
35         logging.info("Copying contents of %s to local", uri)
36 
37         is_local = False
38         if uri.startswith(_LOCAL_PREFIX) or os.path.exists(uri):
39             is_local = True
40 
41         if out_dir is None:
42             if is_local:
43                 # noop if out_dir is not set and the path is local
44                 return Storage._download_local(uri)
45             out_dir = tempfile.mkdtemp()
46 
47         if uri.startswith(_GCS_PREFIX):
48             Storage._download_gcs(uri, out_dir)
49         elif uri.startswith(_S3_PREFIX):
50             Storage._download_s3(uri, out_dir)
51         elif re.search(_BLOB_RE, uri):
52             Storage._download_blob(uri, out_dir)
53         elif is_local:
54             return Storage._download_local(uri, out_dir)
55         else:
56             raise Exception("Cannot recognize storage type for " + uri +
57                             "\n'%s', '%s', and '%s' are the current available storage type." %
58                             (_GCS_PREFIX, _S3_PREFIX, _LOCAL_PREFIX))
59 
60         logging.info("Successfully copied %s to %s", uri, out_dir)
61         return out_dir
62 
63     @staticmethod
64     def _download_s3(uri, temp_dir: str):
65         client = Storage._create_minio_client()
66         bucket_args = uri.replace(_S3_PREFIX, "", 1).split("/", 1)
67         bucket_name = bucket_args[0]
68         bucket_path = bucket_args[1] if len(bucket_args) > 1 else ""
69         objects = client.list_objects(bucket_name, prefix=bucket_path, recursive=True)
70         count = 0
71         for obj in objects:
72             # Replace any prefix from the object key with temp_dir
73             subdir_object_key = obj.object_name.replace(bucket_path, "", 1).strip("/")
74             # fget_object handles directory creation if does not exist
75             if not obj.is_dir:
76                 if subdir_object_key == "":
77                     subdir_object_key = obj.object_name
78                 client.fget_object(bucket_name, obj.object_name,
79                                    os.path.join(temp_dir, subdir_object_key))
80             count = count + 1
81         if count == 0:
82             raise RuntimeError("Failed to fetch model. \
83 The path or model %s does not exist." % (uri))
84 
85     @staticmethod
86     def _download_gcs(uri, temp_dir: str):
87         try:
88             storage_client = storage.Client()
89         except exceptions.DefaultCredentialsError:
90             storage_client = storage.Client.create_anonymous_client()
91         bucket_args = uri.replace(_GCS_PREFIX, "", 1).split("/", 1)
92         bucket_name = bucket_args[0]
93         bucket_path = bucket_args[1] if len(bucket_args) > 1 else ""
94         bucket = storage_client.bucket(bucket_name)
95         prefix = bucket_path
96         if not prefix.endswith("/"):
97             prefix = prefix + "/"
98         blobs = bucket.list_blobs(prefix=prefix)
99         count = 0
100         for blob in blobs:
101             # Replace any prefix from the object key with temp_dir
102             subdir_object_key = blob.name.replace(bucket_path, "", 1).strip("/")
103 
104             # Create necessary subdirectory to store the object locally
105             if "/" in subdir_object_key:
106                 local_object_dir = os.path.join(temp_dir, subdir_object_key.rsplit("/", 1)[0])
107                 if not os.path.isdir(local_object_dir):
108                     os.makedirs(local_object_dir, exist_ok=True)
109             if subdir_object_key.strip() != "":
110                 dest_path = os.path.join(temp_dir, subdir_object_key)
111                 logging.info("Downloading: %s", dest_path)
112                 blob.download_to_filename(dest_path)
113             count = count + 1
114         if count == 0:
115             raise RuntimeError("Failed to fetch model. \
116 The path or model %s does not exist." % (uri))
117 
118     @staticmethod
119     def _download_blob(uri, out_dir: str): # pylint: disable=too-many-locals
120         match = re.search(_BLOB_RE, uri)
121         account_name = match.group(1)
122         storage_url = match.group(2)
123         container_name, prefix = storage_url.split("/", 1)
124 
125         logging.info("Connecting to BLOB account: [%s], container: [%s], prefix: [%s]",
126                      account_name,
127                      container_name,
128                      prefix)
129         try:
130             block_blob_service = BlockBlobService(account_name=account_name)
131             blobs = block_blob_service.list_blobs(container_name, prefix=prefix)
132         except Exception: # pylint: disable=broad-except
133             token = Storage._get_azure_storage_token()
134             if token is None:
135                 logging.warning("Azure credentials not found, retrying anonymous access")
136             block_blob_service = BlockBlobService(account_name=account_name, token_credential=token)
137             blobs = block_blob_service.list_blobs(container_name, prefix=prefix)
138         count = 0
139         for blob in blobs:
140             dest_path = os.path.join(out_dir, blob.name)
141             if "/" in blob.name:
142                 head, tail = os.path.split(blob.name)
143                 if prefix is not None:
144                     head = head[len(prefix):]
145                 if head.startswith('/'):
146                     head = head[1:]
147                 dir_path = os.path.join(out_dir, head)
148                 dest_path = os.path.join(dir_path, tail)
149                 if not os.path.isdir(dir_path):
150                     os.makedirs(dir_path)
151 
152             logging.info("Downloading: %s to %s", blob.name, dest_path)
153             block_blob_service.get_blob_to_path(container_name, blob.name, dest_path)
154             count = count + 1
155         if count == 0:
156             raise RuntimeError("Failed to fetch model. \
157 The path or model %s does not exist." % (uri))
158 
159     @staticmethod
160     def _get_azure_storage_token():
161         tenant_id = os.getenv("AZ_TENANT_ID", "")
162         client_id = os.getenv("AZ_CLIENT_ID", "")
163         client_secret = os.getenv("AZ_CLIENT_SECRET", "")
164         subscription_id = os.getenv("AZ_SUBSCRIPTION_ID", "")
165 
166         if tenant_id == "" or client_id == "" or client_secret == "" or subscription_id == "":
167             return None
168 
169         # note the SP must have "Storage Blob Data Owner" perms for this to work
170         import adal
171         from azure.storage.common import TokenCredential
172 
173         authority_url = "https://login.microsoftonline.com/" + tenant_id
174 
175         context = adal.AuthenticationContext(authority_url)
176 
177         token = context.acquire_token_with_client_credentials(
178             "https://storage.azure.com/",
179             client_id,
180             client_secret)
181 
182         token_credential = TokenCredential(token["accessToken"])
183 
184         logging.info("Retrieved SP token credential for client_id: %s", client_id)
185 
186         return token_credential
187 
188     @staticmethod
189     def _download_local(uri, out_dir=None):
190         local_path = uri.replace(_LOCAL_PREFIX, "", 1)
191         if not os.path.exists(local_path):
192             raise RuntimeError("Local path %s does not exist." % (uri))
193 
194         if out_dir is None:
195             return local_path
196         elif not os.path.isdir(out_dir):
197             os.makedirs(out_dir)
198 
199         if os.path.isdir(local_path):
200             local_path = os.path.join(local_path, "*")
201 
202         for src in glob.glob(local_path):
203             _, tail = os.path.split(src)
204             dest_path = os.path.join(out_dir, tail)
205             logging.info("Linking: %s to %s", src, dest_path)
206             os.symlink(src, dest_path)
207         return out_dir
208 
209     @staticmethod
210     def _create_minio_client():
211         # Adding prefixing "http" in urlparse is necessary for it to be the netloc
212         url = urlparse(os.getenv("AWS_ENDPOINT_URL", "http://s3.amazonaws.com"))
213         use_ssl = url.scheme == 'https' if url.scheme else bool(os.getenv("S3_USE_HTTPS", "true"))
214         return Minio(url.netloc,
215                      access_key=os.getenv("AWS_ACCESS_KEY_ID", ""),
216                      secret_key=os.getenv("AWS_SECRET_ACCESS_KEY", ""),
217                      region=os.getenv("AWS_REGION", ""),
218                      secure=use_ssl)
219
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/python/kfserving/kfserving/storage.py b/python/kfserving/kfserving/storage.py
--- a/python/kfserving/kfserving/storage.py
+++ b/python/kfserving/kfserving/storage.py
@@ -15,9 +15,15 @@
 import glob
 import logging
 import tempfile
+import mimetypes
 import os
 import re
+import shutil
+import tarfile
+import zipfile
+import gzip
 from urllib.parse import urlparse
+import requests 
 from azure.storage.blob import BlockBlobService
 from google.auth import exceptions
 from google.cloud import storage
@@ -27,7 +33,8 @@
 _S3_PREFIX = "s3://"
 _BLOB_RE = "https://(.+?).blob.core.windows.net/(.+)"
 _LOCAL_PREFIX = "file://"
-
+_URI_RE = "https?://(.+)/(.+)"
+_HTTP_PREFIX = "http(s)://"
 
 class Storage(object): # pylint: disable=too-few-public-methods
     @staticmethod
@@ -52,10 +59,12 @@
             Storage._download_blob(uri, out_dir)
         elif is_local:
             return Storage._download_local(uri, out_dir)
+        elif re.search(_URI_RE, uri):
+            return Storage._download_from_uri(uri, out_dir)
         else:
             raise Exception("Cannot recognize storage type for " + uri +
-                            "\n'%s', '%s', and '%s' are the current available storage type." %
-                            (_GCS_PREFIX, _S3_PREFIX, _LOCAL_PREFIX))
+                            "\n'%s', '%s', '%s', and '%s' are the current available storage type." %
+                            (_GCS_PREFIX, _S3_PREFIX, _LOCAL_PREFIX, _HTTP_PREFIX))
 
         logging.info("Successfully copied %s to %s", uri, out_dir)
         return out_dir
@@ -206,6 +215,43 @@
             os.symlink(src, dest_path)
         return out_dir
 
+    @staticmethod
+    def _download_from_uri(uri, out_dir=None):
+        url = urlparse(uri)
+        filename = os.path.basename(url.path)
+        mimetype, encoding = mimetypes.guess_type(uri)
+        local_path = os.path.join(out_dir, filename)
+
+        if filename == '':
+            raise ValueError('No filename contained in URI: %s' % (uri))
+
+        with requests.get(uri, stream=True) as response:
+            if response.status_code != 200:
+                raise RuntimeError("URI: %s returned a %s response code." % (uri, response.status_code))
+            if mimetype == 'application/zip' and not response.headers.get('Content-Type', '').startswith('application/zip'):
+                raise RuntimeError("URI: %s did not respond with \'Content-Type\': \'application/zip\'" % (uri))
+            if mimetype != 'application/zip' and not response.headers.get('Content-Type', '').startswith('application/octet-stream'):
+                raise RuntimeError("URI: %s did not respond with \'Content-Type\': \'application/octet-stream\'" % (uri))
+
+            if encoding == 'gzip':
+                stream = gzip.GzipFile(fileobj=response.raw)
+                local_path = os.path.join(out_dir, f'{filename}.tar')
+            else:
+                stream = response.raw
+            with open(local_path, 'wb') as out:
+                shutil.copyfileobj(stream, out)
+        
+        if mimetype in ["application/x-tar", "application/zip"]:
+            if mimetype == "application/x-tar":
+                archive = tarfile.open(local_path, 'r', encoding='utf-8')
+            else:
+                archive = zipfile.ZipFile(local_path, 'r')
+            archive.extractall(out_dir)
+            archive.close()
+            os.remove(local_path)
+
+        return out_dir
+
     @staticmethod
     def _create_minio_client():
         # Adding prefixing "http" in urlparse is necessary for it to be the netloc

{"golden_diff": "diff --git a/python/kfserving/kfserving/storage.py b/python/kfserving/kfserving/storage.py\n--- a/python/kfserving/kfserving/storage.py\n+++ b/python/kfserving/kfserving/storage.py\n@@ -15,9 +15,15 @@\n import glob\n import logging\n import tempfile\n+import mimetypes\n import os\n import re\n+import shutil\n+import tarfile\n+import zipfile\n+import gzip\n from urllib.parse import urlparse\n+import requests \n from azure.storage.blob import BlockBlobService\n from google.auth import exceptions\n from google.cloud import storage\n@@ -27,7 +33,8 @@\n _S3_PREFIX = \"s3://\"\n _BLOB_RE = \"https://(.+?).blob.core.windows.net/(.+)\"\n _LOCAL_PREFIX = \"file://\"\n-\n+_URI_RE = \"https?://(.+)/(.+)\"\n+_HTTP_PREFIX = \"http(s)://\"\n \n class Storage(object): # pylint: disable=too-few-public-methods\n     @staticmethod\n@@ -52,10 +59,12 @@\n             Storage._download_blob(uri, out_dir)\n         elif is_local:\n             return Storage._download_local(uri, out_dir)\n+        elif re.search(_URI_RE, uri):\n+            return Storage._download_from_uri(uri, out_dir)\n         else:\n             raise Exception(\"Cannot recognize storage type for \" + uri +\n-                            \"\\n'%s', '%s', and '%s' are the current available storage type.\" %\n-                            (_GCS_PREFIX, _S3_PREFIX, _LOCAL_PREFIX))\n+                            \"\\n'%s', '%s', '%s', and '%s' are the current available storage type.\" %\n+                            (_GCS_PREFIX, _S3_PREFIX, _LOCAL_PREFIX, _HTTP_PREFIX))\n \n         logging.info(\"Successfully copied %s to %s\", uri, out_dir)\n         return out_dir\n@@ -206,6 +215,43 @@\n             os.symlink(src, dest_path)\n         return out_dir\n \n+    @staticmethod\n+    def _download_from_uri(uri, out_dir=None):\n+        url = urlparse(uri)\n+        filename = os.path.basename(url.path)\n+        mimetype, encoding = mimetypes.guess_type(uri)\n+        local_path = os.path.join(out_dir, filename)\n+\n+        if filename == '':\n+            raise ValueError('No filename contained in URI: %s' % (uri))\n+\n+        with requests.get(uri, stream=True) as response:\n+            if response.status_code != 200:\n+                raise RuntimeError(\"URI: %s returned a %s response code.\" % (uri, response.status_code))\n+            if mimetype == 'application/zip' and not response.headers.get('Content-Type', '').startswith('application/zip'):\n+                raise RuntimeError(\"URI: %s did not respond with \\'Content-Type\\': \\'application/zip\\'\" % (uri))\n+            if mimetype != 'application/zip' and not response.headers.get('Content-Type', '').startswith('application/octet-stream'):\n+                raise RuntimeError(\"URI: %s did not respond with \\'Content-Type\\': \\'application/octet-stream\\'\" % (uri))\n+\n+            if encoding == 'gzip':\n+                stream = gzip.GzipFile(fileobj=response.raw)\n+                local_path = os.path.join(out_dir, f'{filename}.tar')\n+            else:\n+                stream = response.raw\n+            with open(local_path, 'wb') as out:\n+                shutil.copyfileobj(stream, out)\n+        \n+        if mimetype in [\"application/x-tar\", \"application/zip\"]:\n+            if mimetype == \"application/x-tar\":\n+                archive = tarfile.open(local_path, 'r', encoding='utf-8')\n+            else:\n+                archive = zipfile.ZipFile(local_path, 'r')\n+            archive.extractall(out_dir)\n+            archive.close()\n+            os.remove(local_path)\n+\n+        return out_dir\n+\n     @staticmethod\n     def _create_minio_client():\n         # Adding prefixing \"http\" in urlparse is necessary for it to be the netloc\n", "issue": "Support for general web URI address for hosted KFServing model location\n/kind feature\r\n\r\n**Describe the solution you'd like**\r\n\r\nCurrently there are 4 types of storage URIs that are accepted in KFServing yaml files:\r\n\r\nhttps://github.com/kubeflow/kfserving/blob/e25758f6da020e3e8ad63f0880ab95de6af8395b/python/kfserving/kfserving/storage.py#L26-L30\r\n\r\n(1) GCS\r\n(2) AWS S3\r\n(3) Azure (blob over https://)\r\n(4) local (file://)\r\n\r\nI think it would be useful to be able reference a model file hosted anywhere (via general web URI, aka \"http://\", such as a file on github:\r\n\r\nhttps://github.com/jpatanooga/kubeflow_ops_book_dev/blob/master/kfserving/models/sklearn/iris_sklearn_svm_model.joblib\r\n\r\n(in this example the raw file would be downloaded from the url: https://github.com/jpatanooga/kubeflow_ops_book_dev/blob/master/kfserving/models/sklearn/iris_sklearn_svm_model.joblib?raw=true )\r\n\r\nThis would allow for examples to be executed more efficiently; the KFServing project could even provide a repo of the pre-built models (perhaps), so that example model deployments could be easier to run for new users.\r\n\r\n\r\n\n", "before_files": [{"content": "# Copyright 2020 kubeflow.org.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#    http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport glob\nimport logging\nimport tempfile\nimport os\nimport re\nfrom urllib.parse import urlparse\nfrom azure.storage.blob import BlockBlobService\nfrom google.auth import exceptions\nfrom google.cloud import storage\nfrom minio import Minio\n\n_GCS_PREFIX = \"gs://\"\n_S3_PREFIX = \"s3://\"\n_BLOB_RE = \"https://(.+?).blob.core.windows.net/(.+)\"\n_LOCAL_PREFIX = \"file://\"\n\n\nclass Storage(object): # pylint: disable=too-few-public-methods\n    @staticmethod\n    def download(uri: str, out_dir: str = None) -> str:\n        logging.info(\"Copying contents of %s to local\", uri)\n\n        is_local = False\n        if uri.startswith(_LOCAL_PREFIX) or os.path.exists(uri):\n            is_local = True\n\n        if out_dir is None:\n            if is_local:\n                # noop if out_dir is not set and the path is local\n                return Storage._download_local(uri)\n            out_dir = tempfile.mkdtemp()\n\n        if uri.startswith(_GCS_PREFIX):\n            Storage._download_gcs(uri, out_dir)\n        elif uri.startswith(_S3_PREFIX):\n            Storage._download_s3(uri, out_dir)\n        elif re.search(_BLOB_RE, uri):\n            Storage._download_blob(uri, out_dir)\n        elif is_local:\n            return Storage._download_local(uri, out_dir)\n        else:\n            raise Exception(\"Cannot recognize storage type for \" + uri +\n                            \"\\n'%s', '%s', and '%s' are the current available storage type.\" %\n                            (_GCS_PREFIX, _S3_PREFIX, _LOCAL_PREFIX))\n\n        logging.info(\"Successfully copied %s to %s\", uri, out_dir)\n        return out_dir\n\n    @staticmethod\n    def _download_s3(uri, temp_dir: str):\n        client = Storage._create_minio_client()\n        bucket_args = uri.replace(_S3_PREFIX, \"\", 1).split(\"/\", 1)\n        bucket_name = bucket_args[0]\n        bucket_path = bucket_args[1] if len(bucket_args) > 1 else \"\"\n        objects = client.list_objects(bucket_name, prefix=bucket_path, recursive=True)\n        count = 0\n        for obj in objects:\n            # Replace any prefix from the object key with temp_dir\n            subdir_object_key = obj.object_name.replace(bucket_path, \"\", 1).strip(\"/\")\n            # fget_object handles directory creation if does not exist\n            if not obj.is_dir:\n                if subdir_object_key == \"\":\n                    subdir_object_key = obj.object_name\n                client.fget_object(bucket_name, obj.object_name,\n                                   os.path.join(temp_dir, subdir_object_key))\n            count = count + 1\n        if count == 0:\n            raise RuntimeError(\"Failed to fetch model. \\\nThe path or model %s does not exist.\" % (uri))\n\n    @staticmethod\n    def _download_gcs(uri, temp_dir: str):\n        try:\n            storage_client = storage.Client()\n        except exceptions.DefaultCredentialsError:\n            storage_client = storage.Client.create_anonymous_client()\n        bucket_args = uri.replace(_GCS_PREFIX, \"\", 1).split(\"/\", 1)\n        bucket_name = bucket_args[0]\n        bucket_path = bucket_args[1] if len(bucket_args) > 1 else \"\"\n        bucket = storage_client.bucket(bucket_name)\n        prefix = bucket_path\n        if not prefix.endswith(\"/\"):\n            prefix = prefix + \"/\"\n        blobs = bucket.list_blobs(prefix=prefix)\n        count = 0\n        for blob in blobs:\n            # Replace any prefix from the object key with temp_dir\n            subdir_object_key = blob.name.replace(bucket_path, \"\", 1).strip(\"/\")\n\n            # Create necessary subdirectory to store the object locally\n            if \"/\" in subdir_object_key:\n                local_object_dir = os.path.join(temp_dir, subdir_object_key.rsplit(\"/\", 1)[0])\n                if not os.path.isdir(local_object_dir):\n                    os.makedirs(local_object_dir, exist_ok=True)\n            if subdir_object_key.strip() != \"\":\n                dest_path = os.path.join(temp_dir, subdir_object_key)\n                logging.info(\"Downloading: %s\", dest_path)\n                blob.download_to_filename(dest_path)\n            count = count + 1\n        if count == 0:\n            raise RuntimeError(\"Failed to fetch model. \\\nThe path or model %s does not exist.\" % (uri))\n\n    @staticmethod\n    def _download_blob(uri, out_dir: str): # pylint: disable=too-many-locals\n        match = re.search(_BLOB_RE, uri)\n        account_name = match.group(1)\n        storage_url = match.group(2)\n        container_name, prefix = storage_url.split(\"/\", 1)\n\n        logging.info(\"Connecting to BLOB account: [%s], container: [%s], prefix: [%s]\",\n                     account_name,\n                     container_name,\n                     prefix)\n        try:\n            block_blob_service = BlockBlobService(account_name=account_name)\n            blobs = block_blob_service.list_blobs(container_name, prefix=prefix)\n        except Exception: # pylint: disable=broad-except\n            token = Storage._get_azure_storage_token()\n            if token is None:\n                logging.warning(\"Azure credentials not found, retrying anonymous access\")\n            block_blob_service = BlockBlobService(account_name=account_name, token_credential=token)\n            blobs = block_blob_service.list_blobs(container_name, prefix=prefix)\n        count = 0\n        for blob in blobs:\n            dest_path = os.path.join(out_dir, blob.name)\n            if \"/\" in blob.name:\n                head, tail = os.path.split(blob.name)\n                if prefix is not None:\n                    head = head[len(prefix):]\n                if head.startswith('/'):\n                    head = head[1:]\n                dir_path = os.path.join(out_dir, head)\n                dest_path = os.path.join(dir_path, tail)\n                if not os.path.isdir(dir_path):\n                    os.makedirs(dir_path)\n\n            logging.info(\"Downloading: %s to %s\", blob.name, dest_path)\n            block_blob_service.get_blob_to_path(container_name, blob.name, dest_path)\n            count = count + 1\n        if count == 0:\n            raise RuntimeError(\"Failed to fetch model. \\\nThe path or model %s does not exist.\" % (uri))\n\n    @staticmethod\n    def _get_azure_storage_token():\n        tenant_id = os.getenv(\"AZ_TENANT_ID\", \"\")\n        client_id = os.getenv(\"AZ_CLIENT_ID\", \"\")\n        client_secret = os.getenv(\"AZ_CLIENT_SECRET\", \"\")\n        subscription_id = os.getenv(\"AZ_SUBSCRIPTION_ID\", \"\")\n\n        if tenant_id == \"\" or client_id == \"\" or client_secret == \"\" or subscription_id == \"\":\n            return None\n\n        # note the SP must have \"Storage Blob Data Owner\" perms for this to work\n        import adal\n        from azure.storage.common import TokenCredential\n\n        authority_url = \"https://login.microsoftonline.com/\" + tenant_id\n\n        context = adal.AuthenticationContext(authority_url)\n\n        token = context.acquire_token_with_client_credentials(\n            \"https://storage.azure.com/\",\n            client_id,\n            client_secret)\n\n        token_credential = TokenCredential(token[\"accessToken\"])\n\n        logging.info(\"Retrieved SP token credential for client_id: %s\", client_id)\n\n        return token_credential\n\n    @staticmethod\n    def _download_local(uri, out_dir=None):\n        local_path = uri.replace(_LOCAL_PREFIX, \"\", 1)\n        if not os.path.exists(local_path):\n            raise RuntimeError(\"Local path %s does not exist.\" % (uri))\n\n        if out_dir is None:\n            return local_path\n        elif not os.path.isdir(out_dir):\n            os.makedirs(out_dir)\n\n        if os.path.isdir(local_path):\n            local_path = os.path.join(local_path, \"*\")\n\n        for src in glob.glob(local_path):\n            _, tail = os.path.split(src)\n            dest_path = os.path.join(out_dir, tail)\n            logging.info(\"Linking: %s to %s\", src, dest_path)\n            os.symlink(src, dest_path)\n        return out_dir\n\n    @staticmethod\n    def _create_minio_client():\n        # Adding prefixing \"http\" in urlparse is necessary for it to be the netloc\n        url = urlparse(os.getenv(\"AWS_ENDPOINT_URL\", \"http://s3.amazonaws.com\"))\n        use_ssl = url.scheme == 'https' if url.scheme else bool(os.getenv(\"S3_USE_HTTPS\", \"true\"))\n        return Minio(url.netloc,\n                     access_key=os.getenv(\"AWS_ACCESS_KEY_ID\", \"\"),\n                     secret_key=os.getenv(\"AWS_SECRET_ACCESS_KEY\", \"\"),\n                     region=os.getenv(\"AWS_REGION\", \"\"),\n                     secure=use_ssl)\n", "path": "python/kfserving/kfserving/storage.py"}], "after_files": [{"content": "# Copyright 2020 kubeflow.org.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#    http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport glob\nimport logging\nimport tempfile\nimport mimetypes\nimport os\nimport re\nimport shutil\nimport tarfile\nimport zipfile\nimport gzip\nfrom urllib.parse import urlparse\nimport requests \nfrom azure.storage.blob import BlockBlobService\nfrom google.auth import exceptions\nfrom google.cloud import storage\nfrom minio import Minio\n\n_GCS_PREFIX = \"gs://\"\n_S3_PREFIX = \"s3://\"\n_BLOB_RE = \"https://(.+?).blob.core.windows.net/(.+)\"\n_LOCAL_PREFIX = \"file://\"\n_URI_RE = \"https?://(.+)/(.+)\"\n_HTTP_PREFIX = \"http(s)://\"\n\nclass Storage(object): # pylint: disable=too-few-public-methods\n    @staticmethod\n    def download(uri: str, out_dir: str = None) -> str:\n        logging.info(\"Copying contents of %s to local\", uri)\n\n        is_local = False\n        if uri.startswith(_LOCAL_PREFIX) or os.path.exists(uri):\n            is_local = True\n\n        if out_dir is None:\n            if is_local:\n                # noop if out_dir is not set and the path is local\n                return Storage._download_local(uri)\n            out_dir = tempfile.mkdtemp()\n\n        if uri.startswith(_GCS_PREFIX):\n            Storage._download_gcs(uri, out_dir)\n        elif uri.startswith(_S3_PREFIX):\n            Storage._download_s3(uri, out_dir)\n        elif re.search(_BLOB_RE, uri):\n            Storage._download_blob(uri, out_dir)\n        elif is_local:\n            return Storage._download_local(uri, out_dir)\n        elif re.search(_URI_RE, uri):\n            return Storage._download_from_uri(uri, out_dir)\n        else:\n            raise Exception(\"Cannot recognize storage type for \" + uri +\n                            \"\\n'%s', '%s', '%s', and '%s' are the current available storage type.\" %\n                            (_GCS_PREFIX, _S3_PREFIX, _LOCAL_PREFIX, _HTTP_PREFIX))\n\n        logging.info(\"Successfully copied %s to %s\", uri, out_dir)\n        return out_dir\n\n    @staticmethod\n    def _download_s3(uri, temp_dir: str):\n        client = Storage._create_minio_client()\n        bucket_args = uri.replace(_S3_PREFIX, \"\", 1).split(\"/\", 1)\n        bucket_name = bucket_args[0]\n        bucket_path = bucket_args[1] if len(bucket_args) > 1 else \"\"\n        objects = client.list_objects(bucket_name, prefix=bucket_path, recursive=True)\n        count = 0\n        for obj in objects:\n            # Replace any prefix from the object key with temp_dir\n            subdir_object_key = obj.object_name.replace(bucket_path, \"\", 1).strip(\"/\")\n            # fget_object handles directory creation if does not exist\n            if not obj.is_dir:\n                if subdir_object_key == \"\":\n                    subdir_object_key = obj.object_name\n                client.fget_object(bucket_name, obj.object_name,\n                                   os.path.join(temp_dir, subdir_object_key))\n            count = count + 1\n        if count == 0:\n            raise RuntimeError(\"Failed to fetch model. \\\nThe path or model %s does not exist.\" % (uri))\n\n    @staticmethod\n    def _download_gcs(uri, temp_dir: str):\n        try:\n            storage_client = storage.Client()\n        except exceptions.DefaultCredentialsError:\n            storage_client = storage.Client.create_anonymous_client()\n        bucket_args = uri.replace(_GCS_PREFIX, \"\", 1).split(\"/\", 1)\n        bucket_name = bucket_args[0]\n        bucket_path = bucket_args[1] if len(bucket_args) > 1 else \"\"\n        bucket = storage_client.bucket(bucket_name)\n        prefix = bucket_path\n        if not prefix.endswith(\"/\"):\n            prefix = prefix + \"/\"\n        blobs = bucket.list_blobs(prefix=prefix)\n        count = 0\n        for blob in blobs:\n            # Replace any prefix from the object key with temp_dir\n            subdir_object_key = blob.name.replace(bucket_path, \"\", 1).strip(\"/\")\n\n            # Create necessary subdirectory to store the object locally\n            if \"/\" in subdir_object_key:\n                local_object_dir = os.path.join(temp_dir, subdir_object_key.rsplit(\"/\", 1)[0])\n                if not os.path.isdir(local_object_dir):\n                    os.makedirs(local_object_dir, exist_ok=True)\n            if subdir_object_key.strip() != \"\":\n                dest_path = os.path.join(temp_dir, subdir_object_key)\n                logging.info(\"Downloading: %s\", dest_path)\n                blob.download_to_filename(dest_path)\n            count = count + 1\n        if count == 0:\n            raise RuntimeError(\"Failed to fetch model. \\\nThe path or model %s does not exist.\" % (uri))\n\n    @staticmethod\n    def _download_blob(uri, out_dir: str): # pylint: disable=too-many-locals\n        match = re.search(_BLOB_RE, uri)\n        account_name = match.group(1)\n        storage_url = match.group(2)\n        container_name, prefix = storage_url.split(\"/\", 1)\n\n        logging.info(\"Connecting to BLOB account: [%s], container: [%s], prefix: [%s]\",\n                     account_name,\n                     container_name,\n                     prefix)\n        try:\n            block_blob_service = BlockBlobService(account_name=account_name)\n            blobs = block_blob_service.list_blobs(container_name, prefix=prefix)\n        except Exception: # pylint: disable=broad-except\n            token = Storage._get_azure_storage_token()\n            if token is None:\n                logging.warning(\"Azure credentials not found, retrying anonymous access\")\n            block_blob_service = BlockBlobService(account_name=account_name, token_credential=token)\n            blobs = block_blob_service.list_blobs(container_name, prefix=prefix)\n        count = 0\n        for blob in blobs:\n            dest_path = os.path.join(out_dir, blob.name)\n            if \"/\" in blob.name:\n                head, tail = os.path.split(blob.name)\n                if prefix is not None:\n                    head = head[len(prefix):]\n                if head.startswith('/'):\n                    head = head[1:]\n                dir_path = os.path.join(out_dir, head)\n                dest_path = os.path.join(dir_path, tail)\n                if not os.path.isdir(dir_path):\n                    os.makedirs(dir_path)\n\n            logging.info(\"Downloading: %s to %s\", blob.name, dest_path)\n            block_blob_service.get_blob_to_path(container_name, blob.name, dest_path)\n            count = count + 1\n        if count == 0:\n            raise RuntimeError(\"Failed to fetch model. \\\nThe path or model %s does not exist.\" % (uri))\n\n    @staticmethod\n    def _get_azure_storage_token():\n        tenant_id = os.getenv(\"AZ_TENANT_ID\", \"\")\n        client_id = os.getenv(\"AZ_CLIENT_ID\", \"\")\n        client_secret = os.getenv(\"AZ_CLIENT_SECRET\", \"\")\n        subscription_id = os.getenv(\"AZ_SUBSCRIPTION_ID\", \"\")\n\n        if tenant_id == \"\" or client_id == \"\" or client_secret == \"\" or subscription_id == \"\":\n            return None\n\n        # note the SP must have \"Storage Blob Data Owner\" perms for this to work\n        import adal\n        from azure.storage.common import TokenCredential\n\n        authority_url = \"https://login.microsoftonline.com/\" + tenant_id\n\n        context = adal.AuthenticationContext(authority_url)\n\n        token = context.acquire_token_with_client_credentials(\n            \"https://storage.azure.com/\",\n            client_id,\n            client_secret)\n\n        token_credential = TokenCredential(token[\"accessToken\"])\n\n        logging.info(\"Retrieved SP token credential for client_id: %s\", client_id)\n\n        return token_credential\n\n    @staticmethod\n    def _download_local(uri, out_dir=None):\n        local_path = uri.replace(_LOCAL_PREFIX, \"\", 1)\n        if not os.path.exists(local_path):\n            raise RuntimeError(\"Local path %s does not exist.\" % (uri))\n\n        if out_dir is None:\n            return local_path\n        elif not os.path.isdir(out_dir):\n            os.makedirs(out_dir)\n\n        if os.path.isdir(local_path):\n            local_path = os.path.join(local_path, \"*\")\n\n        for src in glob.glob(local_path):\n            _, tail = os.path.split(src)\n            dest_path = os.path.join(out_dir, tail)\n            logging.info(\"Linking: %s to %s\", src, dest_path)\n            os.symlink(src, dest_path)\n        return out_dir\n\n    @staticmethod\n    def _download_from_uri(uri, out_dir=None):\n        url = urlparse(uri)\n        filename = os.path.basename(url.path)\n        mimetype, encoding = mimetypes.guess_type(uri)\n        local_path = os.path.join(out_dir, filename)\n\n        if filename == '':\n            raise ValueError('No filename contained in URI: %s' % (uri))\n\n        with requests.get(uri, stream=True) as response:\n            if response.status_code != 200:\n                raise RuntimeError(\"URI: %s returned a %s response code.\" % (uri, response.status_code))\n            if mimetype == 'application/zip' and not response.headers.get('Content-Type', '').startswith('application/zip'):\n                raise RuntimeError(\"URI: %s did not respond with \\'Content-Type\\': \\'application/zip\\'\" % (uri))\n            if mimetype != 'application/zip' and not response.headers.get('Content-Type', '').startswith('application/octet-stream'):\n                raise RuntimeError(\"URI: %s did not respond with \\'Content-Type\\': \\'application/octet-stream\\'\" % (uri))\n\n            if encoding == 'gzip':\n                stream = gzip.GzipFile(fileobj=response.raw)\n                local_path = os.path.join(out_dir, f'{filename}.tar')\n            else:\n                stream = response.raw\n            with open(local_path, 'wb') as out:\n                shutil.copyfileobj(stream, out)\n        \n        if mimetype in [\"application/x-tar\", \"application/zip\"]:\n            if mimetype == \"application/x-tar\":\n                archive = tarfile.open(local_path, 'r', encoding='utf-8')\n            else:\n                archive = zipfile.ZipFile(local_path, 'r')\n            archive.extractall(out_dir)\n            archive.close()\n            os.remove(local_path)\n\n        return out_dir\n\n    @staticmethod\n    def _create_minio_client():\n        # Adding prefixing \"http\" in urlparse is necessary for it to be the netloc\n        url = urlparse(os.getenv(\"AWS_ENDPOINT_URL\", \"http://s3.amazonaws.com\"))\n        use_ssl = url.scheme == 'https' if url.scheme else bool(os.getenv(\"S3_USE_HTTPS\", \"true\"))\n        return Minio(url.netloc,\n                     access_key=os.getenv(\"AWS_ACCESS_KEY_ID\", \"\"),\n                     secret_key=os.getenv(\"AWS_SECRET_ACCESS_KEY\", \"\"),\n                     region=os.getenv(\"AWS_REGION\", \"\"),\n                     secure=use_ssl)\n", "path": "python/kfserving/kfserving/storage.py"}]}