Python源码示例：PyQt5.QtNetwork.QNetworkRequest()
示例1
def __init__(self, *args, **kwargs) -> None:
super().__init__(*args, **kwargs)
self._stream_buffer = QByteArray()
self._stream_buffer_start_index = -1
self._network_manager = None # type: QNetworkAccessManager
self._image_request = None # type: QNetworkRequest
self._image_reply = None # type: QNetworkReply
self._image = QImage()
self._image_rect = QRect()
self._source_url = QUrl()
self._started = False
self._mirror = False
self.setAntialiasing(True)
示例2
def __init__(self, *args, **kwargs):
super(OpencvWidget, self).__init__(*args, **kwargs)
self.httpRequestAborted = False
self.fps = 24
self.resize(800, 600)
if not os.path.exists("Data/shape_predictor_68_face_landmarks.dat"):
self.setText("正在下载数据文件。。。")
self.outFile = QFile(
"Data/shape_predictor_68_face_landmarks.dat.bz2")
if not self.outFile.open(QIODevice.WriteOnly):
QMessageBox.critical(self, '错误', '无法写入文件')
return
self.qnam = QNetworkAccessManager(self)
self._reply = self.qnam.get(QNetworkRequest(QUrl(URL)))
self._reply.finished.connect(self.httpFinished)
self._reply.readyRead.connect(self.httpReadyRead)
self._reply.downloadProgress.connect(self.updateDataReadProgress)
else:
self.startCapture()
示例3
def test_dir(self, tmpdir):
url = QUrl.fromLocalFile(str(tmpdir))
req = QNetworkRequest(url)
reply = filescheme.handler(req, None, None)
# The URL will always use /, even on Windows - so we force this here
# too.
tmpdir_path = str(tmpdir).replace(os.sep, '/')
assert reply.readAll() == filescheme.dirbrowser_html(tmpdir_path)
示例4
def test_file(self, tmpdir):
filename = tmpdir / 'foo'
filename.ensure()
url = QUrl.fromLocalFile(str(filename))
req = QNetworkRequest(url)
reply = filescheme.handler(req, None, None)
assert reply is None
示例5
def test_unicode_encode_error(self, mocker):
url = QUrl('file:///tmp/foo')
req = QNetworkRequest(url)
err = UnicodeEncodeError('ascii', '', 0, 2, 'foo')
mocker.patch('os.path.isdir', side_effect=err)
reply = filescheme.handler(req, None, None)
assert reply is None
示例6
def on_download_requested(self, request):
"""Called when the user wants to download a link.
We need to construct a copy of the QNetworkRequest here as the
download_manager needs it async and we'd get a segfault otherwise as
soon as the user has entered the filename, as Qt seems to delete it
after this slot returns.
"""
req = QNetworkRequest(request)
download_manager = objreg.get('qtnetwork-download-manager')
download_manager.get_request(req, qnam=self.networkAccessManager())
示例7
def acceptNavigationRequest(self,
frame: QWebFrame,
request: QNetworkRequest,
typ: QWebPage.NavigationType) -> bool:
"""Override acceptNavigationRequest to handle clicked links.
Setting linkDelegationPolicy to DelegateAllLinks and using a slot bound
to linkClicked won't work correctly, because when in a frameset, we
have no idea in which frame the link should be opened.
Checks if it should open it in a tab (middle-click or control) or not,
and then conditionally opens the URL here or in another tab/window.
"""
type_map = {
QWebPage.NavigationTypeLinkClicked:
usertypes.NavigationRequest.Type.link_clicked,
QWebPage.NavigationTypeFormSubmitted:
usertypes.NavigationRequest.Type.form_submitted,
QWebPage.NavigationTypeFormResubmitted:
usertypes.NavigationRequest.Type.form_resubmitted,
QWebPage.NavigationTypeBackOrForward:
usertypes.NavigationRequest.Type.back_forward,
QWebPage.NavigationTypeReload:
usertypes.NavigationRequest.Type.reloaded,
QWebPage.NavigationTypeOther:
usertypes.NavigationRequest.Type.other,
}
is_main_frame = frame is self.mainFrame()
navigation = usertypes.NavigationRequest(url=request.url(),
navigation_type=type_map[typ],
is_main_frame=is_main_frame)
if navigation.navigation_type == navigation.Type.reloaded:
self.reloading.emit(navigation.url)
self.navigation_request.emit(navigation)
return navigation.accepted
示例8
def requestHook(self, request: QNetworkRequest):
super().requestHook(request)
token = self._account.accessToken
if not self._account.isLoggedIn or token is None:
Logger.warning("Cannot add authorization to Cloud Api request")
return
header_dict = {
"Authorization": "Bearer {}".format(token)
}
self.addHeaders(request, header_dict)
示例9
def start(self) -> None:
self.stop() # Ensure that previous requests (if any) are stopped.
if not self._source_url:
Logger.log("w", "Unable to start camera stream without target!")
return
self._started = True
self._image_request = QNetworkRequest(self._source_url)
if self._network_manager is None:
self._network_manager = QNetworkAccessManager()
self._image_reply = self._network_manager.get(self._image_request)
self._image_reply.downloadProgress.connect(self._onStreamDownloadProgress)
示例10
def requestHook(self, request: QNetworkRequest) -> None:
"""Invoked after request-specific headers are set and before HttpRequestData is created"""
pass
示例11
def addHeaders(request: QNetworkRequest, header_dict: Dict) -> None:
for key, value in header_dict.items():
request.setRawHeader(key.encode("utf-8"), value.encode("utf-8"))
示例12
def requestHook(self, request: QNetworkRequest) -> None:
super().requestHook(request)
self.addHeaders(request, self.header_dict)
示例13
def requestHook(self, request: QNetworkRequest) -> None:
# not calling super().request_hook() because base will do that.
self.base.requestHook(request)
self.addHeaders(request, self.header_dict)
示例14
def make_request(self, url, data, filename):
print(f"Making request to {url}")
# Create the request object
self.request = qtn.QNetworkRequest(url)
# create the multipart object
self.multipart = qtn.QHttpMultiPart(qtn.QHttpMultiPart.FormDataType)
# Write the key-value data to the multipart
for key, value in (data or {}).items():
http_part = qtn.QHttpPart()
http_part.setHeader(
qtn.QNetworkRequest.ContentDispositionHeader,
f'form-data; name="{key}"'
)
http_part.setBody(value.encode('utf-8'))
self.multipart.append(http_part)
# Write the file data to the multipart
if filename:
file_part = qtn.QHttpPart()
filedata = open(filename, 'rb').read()
file_part.setHeader(
qtn.QNetworkRequest.ContentDispositionHeader,
f'form-data; name="attachment"; filename="{filename}"'
)
file_part.setBody(filedata)
self.multipart.append(file_part)
# Post the request with the form data
self.nam.post(self.request, self.multipart)
示例15
def make_request(self, url, data, filename):
print(f"Making request to {url}")
# Create the request object
self.request = qtn.QNetworkRequest(url)
# create the multipart object
self.multipart = qtn.QHttpMultiPart(qtn.QHttpMultiPart.FormDataType)
# Write the key-value data to the multipart
json_string = json.dumps(data)
http_part = qtn.QHttpPart()
http_part.setHeader(
qtn.QNetworkRequest.ContentTypeHeader,
'text/json'
)
http_part.setBody(json_string.encode('utf-8'))
self.multipart.append(http_part)
# Write the file data to the multipart
if filename:
file_part = qtn.QHttpPart()
filedata = open(filename, 'rb').read()
file_part.setHeader(
qtn.QNetworkRequest.ContentDispositionHeader,
f'form-data; name="attachment"; filename="{filename}"'
)
file_part.setBody(filedata)
self.multipart.append(file_part)
# Post the request with the form data
self.nam.post(self.request, self.multipart)
示例16
def __init__(self, url):
super().__init__()
self.manager = qtn.QNetworkAccessManager(
finished=self.on_finished)
self.request = qtn.QNetworkRequest(qtc.QUrl(url))
self.manager.get(self.request)
示例17
def __init__(self, *args, **kwargs) -> None:
super().__init__(*args, **kwargs)
self._stream_buffer = QByteArray()
self._stream_buffer_start_index = -1
self._network_manager = None # type: QNetworkAccessManager
self._image_request = None # type: QNetworkRequest
self._image_reply = None # type: QNetworkReply
self._image = QImage()
self._image_rect = QRect()
self._source_url = QUrl()
self._started = False
self._mirror = False
self.setAntialiasing(True)
## Ensure that close gets called when object is destroyed
示例18
def start(self) -> None:
self.stop() # Ensure that previous requests (if any) are stopped.
if not self._source_url:
Logger.log("w", "Unable to start camera stream without target!")
return
self._started = True
Logger.log("w", "MJPEG starting stream...")
self._image_request = QNetworkRequest(self._source_url)
if self._network_manager is None:
self._network_manager = QNetworkAccessManager()
self._image_reply = self._network_manager.get(self._image_request)
self._image_reply.downloadProgress.connect(self._onStreamDownloadProgress)
示例19
def createRequest(self, op, request, device = None ):
global block_list
try:
urlLnk = (request.url().toString())
except UnicodeEncodeError:
urlLnk = (request.url().path())
if self.get_link:
if self.get_link in urlLnk:
self.netS.emit(urlLnk)
lower_case = urlLnk.lower()
lst = []
if self.default_block:
lst = [
"doubleclick.net", 'adnxs', r"||youtube-nocookie.com/gen_204?",
r"youtube.com###watch-branded-actions", "imagemapurl",
"b.scorecardresearch.com", "rightstuff.com", "scarywater.net",
"popup.js", "banner.htm", "_tribalfusion",
"||n4403ad.doubleclick.net^$third-party",
".googlesyndication.com", "graphics.js", "fonts.googleapis.com/css",
"s0.2mdn.net", "server.cpmstar.com", "||banzai/banner.$subdocument",
"@@||anime-source.com^$document", "/pagead2.", "frugal.gif",
"jriver_banner.png", "show_ads.js",
'##a[href^="http://billing.frugalusenet.com/"]',
"http://jriver.com/video.html", "||animenewsnetwork.com^*.aframe?",
"||contextweb.com^$third-party", ".gutter", ".iab", 'revcontent',
".ads", "ads.", ".bebi", "mgid"
]
if self.block_request:
lst = lst + self.block_request
block = False
for l in lst:
if lower_case.find(l) != -1:
block = True
break
if (self.select_request and self.select_request in urlLnk) or self.print_request:
print(urlLnk)
if block:
return QNetworkAccessManager.createRequest(self, QNetworkAccessManager.GetOperation, QtNetwork.QNetworkRequest(QtCore.QUrl()))
else:
return QNetworkAccessManager.createRequest(self, op, request, device)
示例20
def __init__(self, request_id: str,
http_method: str, request: "QNetworkRequest",
manager_timeout_callback: Callable[["HttpRequestData"], None],
data: Optional[Union[bytes, bytearray]] = None,
callback: Optional[Callable[["QNetworkReply"], None]] = None,
error_callback: Optional[Callable[["QNetworkReply", "QNetworkReply.NetworkError"], None]] = None,
download_progress_callback: Optional[Callable[[int, int], None]] = None,
upload_progress_callback: Optional[Callable[[int, int], None]] = None,
timeout: Optional[float] = None,
reply: Optional["QNetworkReply"] = None,
parent: Optional["QObject"] = None) -> None:
super().__init__(parent = parent)
# Sanity checks
if timeout is not None and timeout <= 0:
raise ValueError("Timeout must be a positive value, but got [%s] instead." % timeout)
self._request_id = request_id
self.http_method = http_method
self.request = request
self.data = data
self.callback = callback
self.error_callback = error_callback
self.download_progress_callback = download_progress_callback
self.upload_progress_callback = upload_progress_callback
self._timeout = timeout
self.reply = reply
# For benchmarking. For calculating the time a request spent pending.
self._create_time = time.time()
# The timestamp when this request was initially issued to the QNetworkManager. This field to used to track and
# manage timeouts (if set) for the requests.
self._start_time = None # type: Optional[float]
self.is_aborted_due_to_timeout = False
self._last_response_time = float(0)
self._timeout_timer = QTimer(parent = self)
if self._timeout is not None:
self._timeout_timer.setSingleShot(True)
timeout_check_interval = self._timeout * 1000 * (1 + self.TIMEOUT_CHECK_TOLERANCE)
self._timeout_timer.setInterval(timeout_check_interval)
self._timeout_timer.timeout.connect(self._onTimeoutTimerTriggered)
self._manager_timeout_callback = manager_timeout_callback
