/
remote.py
213 lines (162 loc) · 5.19 KB
/
remote.py
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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
import io
import requests.utils
from ..exception import FrictionlessException
from ..metadata import Metadata
from ..control import Control
from ..plugin import Plugin
from ..loader import Loader
from .. import errors
from .. import config
# Plugin
class RemotePlugin(Plugin):
"""Plugin for Remote Data
API | Usage
-------- | --------
Public | `from frictionless.plugins.remote import RemotePlugin`
"""
def create_control(self, resource, *, descriptor):
if resource.scheme in config.REMOTE_SCHEMES:
return RemoteControl(descriptor)
def create_loader(self, resource):
if resource.scheme in config.REMOTE_SCHEMES:
return RemoteLoader(resource)
# Control
class RemoteControl(Control):
"""Remote control representation
API | Usage
-------- | --------
Public | `from frictionless.plugins.remote import RemoteControl`
Parameters:
descriptor? (str|dict): descriptor
http_session? (requests.Session): user defined HTTP session
http_preload? (bool): don't use HTTP streaming and preload all the data
http_timeout? (int): user defined HTTP timeout in minutes
Raises:
FrictionlessException: raise any error that occurs during the process
"""
def __init__(
self,
descriptor=None,
*,
http_session=None,
http_preload=None,
http_timeout=None,
newline=None,
detect_encoding=None,
):
self.setinitial("httpSession", http_session)
self.setinitial("httpPreload", http_preload)
self.setinitial("httpTimeout", http_timeout)
super().__init__(descriptor, newline=newline, detect_encoding=detect_encoding)
@Metadata.property
def http_session(self):
"""
Returns:
requests.Session: HTTP session
"""
http_session = self.get("httpSession")
if not http_session:
http_session = requests.Session()
http_session.headers.update(config.DEFAULT_HTTP_HEADERS)
return http_session
@Metadata.property
def http_preload(self):
"""
Returns:
bool: if not streaming
"""
return self.get("httpPreload", False)
@Metadata.property
def http_timeout(self):
"""
Returns:
int: HTTP timeout in minutes
"""
return self.get("httpTimeout", config.DEFAULT_HTTP_TIMEOUT)
# Expand
def expand(self):
"""Expand metadata"""
super().expand()
self.setdefault("httpPreload", self.http_preload)
self.setdefault("httpTimeout", self.http_timeout)
# Metadata
metadata_profile = { # type: ignore
"type": "object",
"additionalProperties": False,
"properties": {
"httpSession": {},
"httpPreload": {"type": "boolean"},
"httpTimeout": {"type": "number"},
"newline": {"type": "string"},
"detectEncoding": {},
},
}
# Loader
class RemoteLoader(Loader):
"""Remote loader implementation.
API | Usage
-------- | --------
Public | `from frictionless.plugins.remote import RemoteLoader`
"""
remote = True
# Read
def read_byte_stream_create(self):
source = requests.utils.requote_uri(self.resource.source)
session = self.resource.control.http_session
timeout = self.resource.control.http_timeout
byte_stream = RemoteByteStream(source, session=session, timeout=timeout).open()
if self.resource.control.http_preload:
buffer = io.BufferedRandom(io.BytesIO())
buffer.write(byte_stream.read())
buffer.seek(0)
byte_stream = buffer
return byte_stream
# Write
def write_byte_stream_save(self, byte_stream):
error = errors.Error(note="Writing to Remote is not supported")
raise FrictionlessException(error)
# Internal
class RemoteByteStream:
def __init__(self, source, *, session, timeout):
self.__source = source
self.__session = session
self.__timeout = timeout
def __iter__(self):
while True:
bytes = self.read(8192)
if not bytes:
break
yield from bytes.splitlines(keepends=True)
def readable(self):
return True
def writable(self):
return False
def seekable(self):
return True
@property
def closed(self):
return self.__closed
def open(self):
self.__closed = False
self.seek(0)
return self
def close(self):
self.__closed = True
def tell(self):
return self.__response.raw.tell()
def flush(self):
pass
def read(self, size=-1):
if size == -1:
size = None
return self.__response.raw.read(size)
def read1(self, size=-1):
return self.read(size)
def seek(self, offset, whence=0):
assert offset == 0
assert whence == 0
self.__response = self.__session.get(
self.__source, stream=True, timeout=self.__timeout
)
self.__response.raise_for_status()
self.__response.raw.decode_content = True