-
-
Notifications
You must be signed in to change notification settings - Fork 24
Expand file tree
/
Copy path__init__.py
More file actions
123 lines (96 loc) · 3.56 KB
/
Copy path__init__.py
File metadata and controls
123 lines (96 loc) · 3.56 KB
1
2
3
4
5
6
7
8
9
10
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
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
# fetchcode is a free software tool from nexB Inc. and others.
# Visit https://github.com/aboutcode-org/fetchcode for support and download.
#
# Copyright (c) nexB Inc. and others. All rights reserved.
# http://nexb.com and http://aboutcode.org
#
# This software is licensed under the Apache License version 2.0.
#
# You may not use this software except in compliance with the License.
# You may obtain a copy of the License at:
# http://apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software distributed
# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
# CONDITIONS OF ANY KIND, either express or implied. See the License for the
# specific language governing permissions and limitations under the License.
import os
import tempfile
from ftplib import FTP
from mimetypes import MimeTypes
from urllib.parse import urlparse
import requests
class Response:
def __init__(self, location, content_type, size, url):
"""
Represent the response from fetching a URL with:
- `location`: the absolute location of the files that was fetched
- `content_type`: content type of the file
- `size`: size of the retrieved content in bytes
- `url`: fetched URL
"""
self.url = url
self.size = size
self.content_type = content_type
self.location = location
def fetch_http(url, location):
"""
Return a `Response` object built from fetching the content at a HTTP/HTTPS based
`url` URL string saving the content in a file at `location`
"""
r = requests.get(url)
with open(location, "wb") as f:
f.write(r.content)
content_type = r.headers.get("content-type")
size = r.headers.get("content-length")
size = int(size) if size else None
resp = Response(location=location, content_type=content_type, size=size, url=url)
return resp
def fetch_ftp(url, location):
"""
Return a `Response` object built from fetching the content at a FTP based `url` URL string
saving the content in a file at `location`
"""
url_parts = urlparse(url)
netloc = url_parts.netloc
path = url_parts.path
dir, file = os.path.split(path)
ftp = FTP(netloc)
ftp.login()
size = ftp.size(path)
mime = MimeTypes()
mime_type = mime.guess_type(file)
if mime_type:
content_type = mime_type[0]
else:
content_type = None
ftp.cwd(dir)
file = "RETR {}".format(file)
with open(location, "wb") as f:
ftp.retrbinary(file, f.write)
ftp.close()
resp = Response(location=location, content_type=content_type, size=size, url=url)
return resp
def fetch(url):
"""
Return a `Response` object built from fetching the content at the `url` URL string and
store content at a temporary file.
"""
temp = tempfile.NamedTemporaryFile(delete=False)
location = temp.name
url_parts = urlparse(url)
scheme = url_parts.scheme
fetchers = {"ftp": fetch_ftp, "http": fetch_http, "https": fetch_http}
if scheme in fetchers:
return fetchers.get(scheme)(url, location)
raise Exception("Not a supported/known scheme.")
def fetch_json_response(url):
"""
Fetch a JSON response from the given URL and return the parsed JSON data.
"""
response = requests.get(url)
if response.status_code != 200:
raise Exception(f"Failed to fetch {url}: {response.status_code} {response.reason}")
try:
return response.json()
except ValueError as e:
raise Exception(f"Failed to parse JSON from {url}: {str(e)}")