Xqt has submitted this change. (
https://gerrit.wikimedia.org/r/c/pywikibot/core/+/983213?usp=email )
Change subject: sse: use requests_sse instead of unsupported sseclient for
EventStreams
......................................................................
sse: use requests_sse instead of unsupported sseclient for EventStreams
Bug: T309380
Change-Id: Ibc6675eed57d939b2bfb884f55ecdd735c7e624b
---
M .github/workflows/doctest.yml
M pywikibot/comms/eventstreams.py
M requirements.txt
M setup.py
M tests/eventstreams_tests.py
M tests/pagegenerators_tests.py
6 files changed, 105 insertions(+), 78 deletions(-)
Approvals:
Xqt: Verified; Looks good to me, approved
diff --git a/.github/workflows/doctest.yml b/.github/workflows/doctest.yml
index 4898c46..ff11951 100644
--- a/.github/workflows/doctest.yml
+++ b/.github/workflows/doctest.yml
@@ -58,7 +58,7 @@
pip install packaging
pip install "PyMySQL >= 1.0.0"
pip install pytest
- pip install "sseclient<0.0.23,>=0.0.18"
+ pip install requests-sse
pip install wikitextparser
- name: Generate user files
diff --git a/pywikibot/comms/eventstreams.py b/pywikibot/comms/eventstreams.py
index 8dd2daa..2e91566 100644
--- a/pywikibot/comms/eventstreams.py
+++ b/pywikibot/comms/eventstreams.py
@@ -2,11 +2,13 @@
This file is part of the Pywikibot framework.
-This module requires sseclient to be installed::
+This module requires requests-sse to be installed::
- pip install "sseclient<0.0.23,>=0.0.18"
+ pip install "requests-sse>=0.5.0"
.. versionadded:: 3.0
+.. versionchanged:: 10.0
+ ``requests-sse`` package is required instead of ``sseclient``.
"""
#
# (C) Pywikibot team, 2017-2025
@@ -16,23 +18,32 @@
from __future__ import annotations
import json
+from datetime import timedelta
from functools import partial
+from typing import Any
from requests.packages.urllib3.exceptions import ProtocolError
from requests.packages.urllib3.util.response import httplib
from pywikibot import Site, Timestamp, config, debug, warning
from pywikibot.backports import NoneType
-from pywikibot.tools import cached
+from pywikibot.tools import cached, deprecated_args
from pywikibot.tools.collections import GeneratorWrapper
try:
- from sseclient import SSEClient as EventSource
-except ImportError as e:
+ from requests_sse import EventSource
+except ModuleNotFoundError as e:
EventSource = e
+INSTALL_MSG = """requests-sse is required for EventStreams;
+install it with
+
+ pip install "requests-sse>=0.5.0"
+"""
+
+
class EventStreams(GeneratorWrapper):
"""Generator class for Server-Sent Events (SSE) protocol.
@@ -99,47 +110,76 @@
>>> del stream
.. versionchanged:: 7.6
- subclassed from :class:`tools.collections.GeneratorWrapper`
+ subclassed from :class:`tools.collections.GeneratorWrapper`.
+ .. versionchanged:: 10.0
+ *retry* value is doubled for each consecutive connect try.
"""
+ @deprecated_args(last_id='last_event_id') # since 10.0.0
def __init__(self, **kwargs) -> None:
"""Initializer.
:keyword bool canary: if True, include canary events, see
- https://w.wiki/7$2z for more info
- :keyword APISite site: a project site object. Used if no url is
- given
- :keyword pywikibot.Timestamp or str since: a timestamp for older
+ https://w.wiki/7$2z for more info.
+ :keyword APISite site: a project site object. Used if no *url*
+ is given.
+ :keyword int retry: Number of milliseconds to wait after disconnects
+ before attempting to reconnect. The server may change this
+ by including a 'retry' line in a message. Retries are handled
+ automatically.
+
+ .. versionchanged:: 10.0
+ 5 seconds are used instead of 3 seconds as default.
+
+ :keyword pywikibot.Timestamp | str since: a timestamp for older
events; there will likely be between 7 and 31 days of
history available but is not guaranteed. It may be given as
a pywikibot.Timestamp, an ISO 8601 string or a mediawiki
timestamp string.
- :keyword Iterable[str] or str streams: event stream types.
+ :keyword Iterable[str] | str streams: event stream types.
Mandatory when no url is given. Multiple streams may be
given as a string with comma separated stream types or an
iterable of strings
- :keyword int or float or tuple[int or float, int or float] timeout:
+ :keyword int | float | tuple[int | float, int | float] timeout:
a timeout value indication how long to wait to send data
before giving up
:keyword str url: an url retrieving events from. Will be set up
to a default url using _site.family settings, stream types
and timestamp
- :param kwargs: keyword arguments passed to `SSEClient` and
- `requests` library
- :raises ImportError: sseclient is not installed
+
+ :keyword Any last_event_id: [*requests-sse*] If provided, this
+ parameter will be sent to the server to tell it to return
+ only messages more recent than this ID.
+ :keyword requests.Session session: [*requests-sse*] specifies a
+ requests.Session, if not, create a default requests.Session.
+ :keyword Callable[[], None] on_open: [*requests-sse*] event
+ handler for open event
+ :keyword Callable[[requests_sse.MessageEvent], None] on_message:
+ [*requests-sse*] event handler for message event
+ :keyword Callable[[], None] on_error: [*requests-sse*] event
+ handler for error event
+ :keyword int chunk_size: [*requests*] A maximum size of the chunk
+ for chunk-encoded requests.
+
+ .. versionchanged:: 10.0
+ None is used instead of 1024 as default value.
+
+ :param kwargs: Other keyword arguments passed to `requests_sse`
+ and `requests` library
+ :raises ModuleNotFoundError: requests-sse is not installed
:raises NotImplementedError: no stream types specified
.. seealso:: https://stream.wikimedia.org/?doc#streams for
available Wikimedia stream types to be passed with `streams`
parameter.
+ .. note:: *retry* keyword argument is used instead of the
+ underlying *reconnection_time* argument which is ignored.
"""
- if isinstance(EventSource, Exception):
- raise ImportError(
- 'sseclient is required for EventStreams;\n'
- 'install it with "pip install sseclient==0.0.22"\n'
- )
+ if isinstance(EventSource, ModuleNotFoundError):
+ raise ImportError(INSTALL_MSG) from EventSource
+
self.filter = {'all': [], 'any': [], 'none': []}
- self._total = None
+ self._total: int | None = None
self._canary = kwargs.pop('canary', False)
try:
@@ -161,6 +201,11 @@
self._url = kwargs.get('url') or self.url
kwargs.setdefault('url', self._url)
+
+ retry = kwargs.pop('retry', None)
+ if retry:
+ kwargs['reconnection_time'] = timedelta(milliseconds=retry)
+
kwargs.setdefault('timeout', config.socket_timeout)
self.sse_kwargs = kwargs
@@ -176,12 +221,12 @@
kwargs['since'] = self._since
if kwargs['timeout'] == config.socket_timeout:
kwargs.pop('timeout')
- return '{}({})'.format(self.__class__.__name__, ', '.join(
+ return '{}({})'.format(type(self).__name__, ', '.join(
f'{k}={v!r}' for k, v in kwargs.items()))
@property
@cached
- def url(self):
+ def url(self) -> str:
"""Get the EventStream's url.
:raises NotImplementedError: no stream types specified
@@ -195,7 +240,7 @@
streams=self._streams,
since=f'?since={self._since}' if self._since else '')
- def set_maximum_items(self, value: int) -> None:
+ def set_maximum_items(self, value: int | None) -> None:
"""Set the maximum number of items to be retrieved from the stream.
If not called, most queries will continue as long as there is
@@ -209,7 +254,7 @@
debug(f'{type(self).__name__}: Set limit (maximum_items) to '
f'{self._total}.')
- def register_filter(self, *args, **kwargs):
+ def register_filter(self, *args, **kwargs) -> None:
"""Register a filter.
Filter types:
@@ -252,6 +297,7 @@
register_filter(ftype='none', bot=True) # 3
Explanation for the result of the filter function:
+
1. ``return data['sever_name'] == 'de.wikipedia.org'``
2. ``return data['type'] in ('edit', 'log')``
3. ``return data['bot'] is True``
@@ -297,7 +343,7 @@
else:
self.filter[ftype].append(partial(_in, key=key, value=value))
- def streamfilter(self, data: dict):
+ def streamfilter(self, data: dict[str, Any]) -> bool:
"""Filter function for eventstreams.
See the description of register_filter() how it works.
@@ -309,10 +355,13 @@
if any(function(data) for function in self.filter['none']):
return False
+
if not all(function(data) for function in self.filter['all']):
return False
+
if not self.filter['any']:
return True
+
return any(function(data) for function in self.filter['any'])
@property
@@ -327,24 +376,20 @@
while self._total is None or n < self._total:
if not hasattr(self, 'source'):
self.source = EventSource(**self.sse_kwargs)
- # sseclient >= 0.0.18 is required for eventstreams (T184713)
- # we don't have a version string inside but the instance
- # variable 'chunk_size' was newly introduced with 0.0.18
- if not hasattr(self.source, 'chunk_size'):
- warning(
- 'You may not have the right sseclient version;\n'
- 'sseclient >= 0.0.18 is required for eventstreams.\n'
- "Install it with 'pip install \"sseclient>=0.0.18\"'")
+ self.source.connect(config.max_retries)
+
try:
event = next(self.source)
except (ProtocolError, OSError, httplib.IncompleteRead) as e:
warning(
f'Connection error: {e}.\nTry to re-establish connection.')
+ self.source.close()
del self.source
if event is not None:
- self.sse_kwargs['last_id'] = event.id
+ self.sse_kwargs['last_event_id'] = event.last_event_id
continue
- if event.event == 'message':
+
+ if event.type == 'message':
if event.data:
try:
element = json.loads(event.data)
@@ -355,13 +400,15 @@
n += 1
yield element
# else: ignore empty message
- elif event.event == 'error':
+ elif event.type == 'error':
warning(f'Encountered error: {event.data}')
else:
- warning(f'Unknown event {event.event} occurred.')
+ warning(f'Unknown event {event.type} occurred.')
debug(f'{type(self).__name__}: Stopped iterating due to exceeding item'
' limit.')
+
+ self.source.close()
del self.source
@@ -373,11 +420,10 @@
:param total: the maximum number of changes to return
:return: pywikibot.comms.eventstream.rc_listener configured for given site
- :raises ImportError: sseclient installation is required
+ :raises ModuleNotFoundError: requests-sse installation is required
"""
- if isinstance(EventSource, Exception):
- raise ImportError('sseclient is required for EventStreams;\n'
- 'install it with "pip install sseclient"\n')
+ if isinstance(EventSource, ModuleNotFoundError):
+ raise ModuleNotFoundError(INSTALL_MSG) from EventSource
stream = EventStreams(streams='recentchange', site=site)
stream.set_maximum_items(total)
diff --git a/requirements.txt b/requirements.txt
index 7a771d3..30a0c0b 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -47,7 +47,7 @@
# core pagegenerators
google >= 1.7
-sseclient >= 0.0.18,< 0.0.23
+requests-sse >= 0.5.0
# The mysql generator in pagegenerators depends on PyMySQL
PyMySQL >= 1.0.0
diff --git a/setup.py b/setup.py
index 942bc94..31f1ec5 100755
--- a/setup.py
+++ b/setup.py
@@ -20,7 +20,7 @@
.. warning:: do not upload a development release to pypi.
"""
#
-# (C) Pywikibot team, 2009-2024
+# (C) Pywikibot team, 2009-2025
#
# Distributed under the terms of the MIT license.
#
@@ -37,7 +37,7 @@
# ------- setup extra_requires ------- #
extra_deps = {
# Core library dependencies
- 'eventstreams': ['sseclient<0.0.23,>=0.0.18'], # T222885
+ 'eventstreams': ['requests-sse>=0.5.0'],
'isbn': ['python-stdnum>=1.19'],
'Graphviz': ['pydot>=1.4.1'],
'Google': ['google>=1.7'],
diff --git a/tests/eventstreams_tests.py b/tests/eventstreams_tests.py
index ea8d7e3..c95e81d 100755
--- a/tests/eventstreams_tests.py
+++ b/tests/eventstreams_tests.py
@@ -1,7 +1,7 @@
#!/usr/bin/env python3
"""Tests for the eventstreams module."""
#
-# (C) Pywikibot team, 2017-2024
+# (C) Pywikibot team, 2017-2025
#
# Distributed under the terms of the MIT license.
#
@@ -272,25 +272,18 @@
while self._total is None or n < self._total:
if not hasattr(self, 'source'):
self.source = EventSource(**self.sse_kwargs)
+ self.source.connect()
+
event = next(self.source)
- if event.event == 'message':
- if not event.data:
- continue
+ if event.type == 'message' and event.data:
n += 1
- try:
- element = json.loads(event.data)
- except ValueError as e: # pragma: no cover
- self.source.resp.close() # close SSLSocket
- del self.source
- raise ValueError(
- f'{e}\n\nEvent no {n}: '
- f'Could not load json data from source\n${event}$'
- ) from e
- yield element
+ yield json.loads(event.data)
+
+ self.source.close()
del self.source
-@require_modules('sseclient')
+@require_modules('requests_sse')
class TestEventSource(TestCase):
"""Test sseclient.EventSource."""
@@ -298,12 +291,7 @@
net = True
def test_stream(self):
- """Verify that the EventSource delivers events without problems.
-
- As found in sseclient 0.0.24 the EventSource gives randomly a
- ValueError 'Unterminated string' when json.load is processed
- if the limit is high enough.
- """
+ """Verify that the EventSource delivers events without problems."""
with skipping(NotImplementedError):
self.es = EventStreamsTestClass(streams='recentchange')
limit = 50
diff --git a/tests/pagegenerators_tests.py b/tests/pagegenerators_tests.py
index d72c392..2fdcbbd 100755
--- a/tests/pagegenerators_tests.py
+++ b/tests/pagegenerators_tests.py
@@ -1,7 +1,7 @@
#!/usr/bin/env python3
"""Test pagegenerators module."""
#
-# (C) Pywikibot team, 2009-2024
+# (C) Pywikibot team, 2009-2025
#
# Distributed under the terms of the MIT license.
from __future__ import annotations
@@ -27,7 +27,6 @@
PreloadingGenerator,
WikibaseItemFilterPageGenerator,
)
-from pywikibot.tools import has_module
from tests import join_data_path, unittest_print
from tests.aspects import (
DefaultSiteTestCase,
@@ -35,6 +34,7 @@
RecentChangesTestCase,
TestCase,
WikidataTestCase,
+ require_modules,
)
from tests.tools_tests import GeneratorIntersectTestCase
from tests.utils import skipping
@@ -1648,17 +1648,10 @@
"""Test case for Live Recent Changes pagegenerator."""
- @classmethod
- def setUpClass(cls):
- """Setup test class."""
- super().setUpClass()
- cls.client = 'sseclient'
- if not has_module(cls.client):
- raise unittest.SkipTest(f'{cls.client} is not available')
-
+ @require_modules('requests_sse')
def test_RC_pagegenerator_result(self):
"""Test RC pagegenerator."""
- lgr = logging.getLogger(self.client)
+ lgr = logging.getLogger('requests_sse.client')
lgr.setLevel(logging.DEBUG)
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
--
To view, visit
https://gerrit.wikimedia.org/r/c/pywikibot/core/+/983213?usp=email
To unsubscribe, or for help writing mail filters, visit
https://gerrit.wikimedia.org/r/settings?usp=email
Gerrit-MessageType: merged
Gerrit-Project: pywikibot/core
Gerrit-Branch: master
Gerrit-Change-Id: Ibc6675eed57d939b2bfb884f55ecdd735c7e624b
Gerrit-Change-Number: 983213
Gerrit-PatchSet: 30
Gerrit-Owner: Xqt <[email protected]>
Gerrit-Reviewer: Xqt <[email protected]>
Gerrit-Reviewer: jenkins-bot
_______________________________________________
Pywikibot-commits mailing list -- [email protected]
To unsubscribe send an email to [email protected]
