diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d86b813..9be07e2 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -32,7 +32,7 @@ jobs: runs-on: ubuntu-latest strategy: matrix: - py-version: ['3.8', '3.9', '3.10'] + py-version: ['3.8', '3.10', '3.13', '3.14'] services: redis-server: image: redis diff --git a/CHANGELOG.md b/CHANGELOG.md index e19de64..1f10f8f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # CHANGELOG +## [4.1.0] - 2025-12-18 +### Changed +- Update grpcio-tools requirements +- Remove deprecated multiprocessing server implementation and CLI options +- Use importlib.metadata for Python 3.8+, fallback to pkg_resources for older versions + ## [4.0.0] - 2025-08-28 ### Changed - Update protobuf requirement diff --git a/requirements.txt b/requirements.txt index 0c0a83b..bf84332 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,7 +1,7 @@ -grpcio>=1.49.0,<1.69.0 -grpcio-tools>=1.49.0,<1.69.0 +grpcio>=1.49.0,<1.77.0 +grpcio-tools>=1.49.0,<1.77.0 grpcio-reflection -protobuf<=6.31.1 +protobuf<=6.34.0 pendulum blinker<=1.8.2 jinja2==3.1.2 diff --git a/sea/app.py b/sea/app.py index 2956356..b53ca73 100644 --- a/sea/app.py +++ b/sea/app.py @@ -29,8 +29,6 @@ class BaseApp: "TESTING": False, "TIMEZONE": "UTC", "GRPC_WORKERS": 4, - "GRPC_THREADS": 1, # Only appliable in multiprocessing server - "GRPC_WORKER_MODE": "threading", # Worker mode. threading|multiprocessing "GRPC_HOST": "0.0.0.0", "GRPC_PORT": 6000, "GRPC_LOG_LEVEL": "WARNING", diff --git a/sea/cli.py b/sea/cli.py index 6a3814c..6a983c2 100644 --- a/sea/cli.py +++ b/sea/cli.py @@ -1,12 +1,26 @@ import sys import os import argparse -import pkg_resources import logging from sea.utils import import_string from sea import create_app +# Use importlib.metadata for Python 3.8+, fallback to pkg_resources for older versions +try: + from importlib.metadata import entry_points +except ImportError: + # Python < 3.8 + try: + import pkg_resources + + def entry_points(group=None): + return pkg_resources.iter_entry_points(group) + except ImportError: + # Fallback if neither is available + def entry_points(group=None): + return [] + class JobException(RuntimeError): pass @@ -60,12 +74,25 @@ def _load_jobs(): import_string("sea.cmds") # load lib jobs - for ep in pkg_resources.iter_entry_points("sea.jobs"): + try: + # Try Python 3.10+ API first: entry_points(group="...") try: - ep.load() - except Exception as e: - logger = logging.getLogger("sea.cmd") - logger.error("error has occurred during pkg loading: {}".format(e)) + eps = entry_points(group="sea.jobs") + except TypeError: + # Python 3.8-3.9: entry_points() returns a dict-like object + all_eps = entry_points() + eps = all_eps.get("sea.jobs", []) + + # Iterate over entry points (works for both EntryPoints object and list/iterator) + for ep in eps: + try: + ep.load() + except Exception as e: + logger = logging.getLogger("sea.cmd") + logger.error("error has occurred during pkg loading: {}".format(e)) + except Exception as e: + logger = logging.getLogger("sea.cmd") + logger.warning("Failed to load entry points: {}".format(e)) # load app jobs appjobs = os.path.join(path, "jobs") diff --git a/sea/cmds.py b/sea/cmds.py index 1654dc5..c1df35a 100644 --- a/sea/cmds.py +++ b/sea/cmds.py @@ -5,23 +5,10 @@ @jobm.job("server", aliases=["s"], help="Run Server") -@jobm.option( - "-M", - "--worker_mode", - required=False, - action="store", - help="Worker mode. threading|multiprocessing", -) -def server(worker_mode): - worker_mode = worker_mode or current_app.config["GRPC_WORKER_MODE"] - if worker_mode == "threading": - from sea.server.threading import Server - - s = Server(current_app) - else: - from sea.server.multiprocessing import Server +def server(): + from sea.server import Server - s = Server(current_app) + s = Server(current_app) s.run() return 0 diff --git a/sea/server/multiprocessing.py b/sea/server/multiprocessing.py deleted file mode 100644 index 7b491b3..0000000 --- a/sea/server/multiprocessing.py +++ /dev/null @@ -1,183 +0,0 @@ -import contextlib -import glob -import multiprocessing -import os -import signal -import socket -import time -from concurrent import futures -from typing import List - -import grpc -from grpc_reflection.v1alpha import reflection - -from sea import signals - - -class Server: - """sea multiprocessing server implementation - - :param app: application instance - """ - - def __init__(self, app): - # application instance - self.app = app - # worker process number - self.worker_num: int = self.app.config["GRPC_WORKERS"] - # worker thread number - self.thread_num: int = self.app.config.get("GRPC_THREADS") - self.host: str = self.app.config["GRPC_HOST"] - self.port: int = self.app.config["GRPC_PORT"] - # slave worker refs, master node contains all slave workers refs - self.workers: List[multiprocessing.Process] = [] - self._stopped: bool = False - # slave worker server instance ref - self.server: grpc.Server = None - - def _run_server(self, bind_address): - server = grpc.server( - futures.ThreadPoolExecutor(max_workers=self.thread_num), - options=[ - ( - "grpc.so_reuseport", - 1, - ), # multiprocessing worker must reuse port to pass between processes. - ], - ) - self.server = server # set server in slave process - # register reflection service - if self.app.config.get("GRPC_REFLECTION_SERVICES"): - reflection.enable_server_reflection((reflection.SERVICE_NAME, *self.app.config["GRPC_REFLECTION_SERVICES"]), self.server) - - for _, (add_func, servicer) in self.app.servicers.items(): - add_func(servicer(), server) - server.add_insecure_port(bind_address) - server.start() - - signals.server_started.send(self) - - # hang up here, to make slave run always - server.wait_for_termination() - - def _run_prometheus_http_server(self): - """Run prometheus_client built-in http server. - - Duing to prometheus_client multiprocessing details, - PROMETHEUS_MULTIPROC_DIR must set in environment variables.""" - if not self.app.config["PROMETHEUS_SCRAPE"]: - return - - from prometheus_client import REGISTRY, start_http_server - from prometheus_client.multiprocess import MultiProcessCollector - - MultiProcessCollector(REGISTRY) - start_http_server(self.app.config["PROMETHEUS_PORT"]) - - def _clean_prometheus(self): - if not self.app.config["PROMETHEUS_SCRAPE"]: - return - dir = os.getenv("PROMETHEUS_MULTIPROC_DIR") - self.app.logger.info(f"clean prometheus dir {dir}") - for f in glob.glob(os.path.join(dir, "*")): - os.remove(f) - - def run(self): - self._run_prometheus_http_server() - - self._register_signals() - - with _reserve_address_port(self.host, self.port) as bind_address: - for _ in range(self.worker_num): - worker = multiprocessing.Process( - target=self._run_server, args=(bind_address,) - ) - worker.start() - self.workers.append(worker) - for worker in self.workers: - worker.join() - - self._clean_prometheus() - - return True - - def _register_signals(self): - signal.signal(signal.SIGINT, self._stop_handler) - signal.signal(signal.SIGHUP, self._stop_handler) - signal.signal(signal.SIGTERM, self._stop_handler) - signal.signal(signal.SIGQUIT, self._stop_handler) - - def _stop_handler(self, signum, frame): - grace = max(self.app.config.get("GRPC_GRACE", 0), 5) - - if self._stopped: - self.app.logger.debug( - "stop signal has received, ignore duplicated function signal" - ) - return - self._stopped = True - - if not self.server: - # master - self.app.logger.warning( - "master process received signal {}, sleep {} to wait slave done".format( - signum, grace - ) - ) - signals.server_stopped.send(self) - - # master process sleep to wait slaves end their lives - time.sleep(grace) - - # kill the slave process which don't wanna die - for worker in self.workers: - if worker.is_alive(): - self.app.logger.warning( - "master found process {} still alive after {} timeout".format( - worker.pid, grace - ) - ) - # compatitable with 3.6 and before - if callable(getattr(worker, "kill", None)): - worker.kill() - else: - os.kill(worker.pid, signal.SIGKILL) - self.app.logger.warning("master exit") - else: - # slave - signals.server_stopped.send(self) - self.app.logger.warning( - "slave process received signal {}, try to stop process".format(signum) - ) - # slave process sleep less 3s to make grace more reliable - self.server.stop(grace - 3) - time.sleep(grace - 3) - - -@contextlib.contextmanager -def _reserve_address_port(host, port): - """Find and reserve a port for all subprocesses to use.""" - - from ipaddress import IPv6Address, ip_address - - ipv6 = False - if host and type(ip_address(host)) is IPv6Address: - ipv6 = True - - sock = socket.socket( - socket.AF_INET6 if ipv6 else socket.AF_INET, socket.SOCK_STREAM - ) - - # ENABLE SO_REUSEPORT - sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) - if sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT) == 0: - raise RuntimeError("Failed to set SO_REUSEPORT.") - - sock.bind((host, port)) - try: - if ipv6: - yield "[{0}]:{1}".format(*sock.getsockname()) - else: - yield "{0}:{1}".format(*sock.getsockname()) - finally: - sock.close() diff --git a/tests/test_cmds.py b/tests/test_cmds.py index 4183e08..1545ecd 100644 --- a/tests/test_cmds.py +++ b/tests/test_cmds.py @@ -10,12 +10,7 @@ def test_cmd_server(app): sys.argv = "sea s".split() - with mock.patch("sea.server.threading.Server", autospec=True) as mocked: - assert cli.main() == 0 - mocked.return_value.run.assert_called_with() - - sys.argv = "sea s -M multiprocessing".split() - with mock.patch("sea.server.multiprocessing.Server", autospec=True) as mocked: + with mock.patch("sea.server.Server", autospec=True) as mocked: assert cli.main() == 0 mocked.return_value.run.assert_called_with() @@ -118,13 +113,24 @@ class FailedEntryPoint: def load(self): raise Exception("Failed entry point") - def new_entry_iter(name): - return [EntryPoint(), FailedEntryPoint()] + # Mock entry_points to return our entry points + class MockEntryPoints: + def __init__(self, entries): + self.entries = entries + def __iter__(self): + return iter(self.entries) + + def mock_entry_points(group=None): + if group == "sea.jobs": + return MockEntryPoints([EntryPoint(), FailedEntryPoint()]) + return MockEntryPoints([]) mock_logger = mock.Mock() - with mock.patch( - "pkg_resources.iter_entry_points", new=new_entry_iter - ), mock.patch("logging.getLogger", return_value=mock_logger): + with mock.patch("sea.cli.entry_points", side_effect=mock_entry_points), \ + mock.patch("logging.getLogger", return_value=mock_logger), \ + mock.patch("os.getcwd", return_value=app.root_path): + # Reload jobs to pick up the mocked entry points + cli._load_jobs() sys.argv = "sea xyz".split() assert cli.main() is None assert app.config.get("XYZ") == "hello" @@ -133,6 +139,98 @@ def new_entry_iter(name): ) +def test_cmd_job_importlib_metadata_python310(app): + """Test entry points loading with importlib.metadata (Python 3.10+ API)""" + class EntryPoint: + def load(self): + @cli.jobm.job("test_importlib") + def f(): + app.config["IMPORTLIB_TEST"] = "success" + return f + + class FailedEntryPoint: + def load(self): + raise Exception("Failed entry point") + + # Mock Python 3.10+ API: entry_points(group="...") returns EntryPoints object + class MockEntryPoints: + def __init__(self, entries): + self.entries = entries + + def __iter__(self): + return iter(self.entries) + + def mock_entry_points(group=None): + if group == "sea.jobs": + return MockEntryPoints([EntryPoint(), FailedEntryPoint()]) + return MockEntryPoints([]) + + mock_logger = mock.Mock() + with mock.patch("sea.cli.entry_points", side_effect=mock_entry_points), \ + mock.patch("logging.getLogger", return_value=mock_logger), \ + mock.patch("os.getcwd", return_value=app.root_path): + # Reload jobs to pick up the mocked entry points + cli._load_jobs() + sys.argv = "sea test_importlib".split() + assert cli.main() is None + assert app.config.get("IMPORTLIB_TEST") == "success" + mock_logger.error.assert_called_with( + "error has occurred during pkg loading: Failed entry point" + ) + + +def test_cmd_job_importlib_metadata_python38(app): + """Test entry points loading with importlib.metadata (Python 3.8-3.9 API)""" + class EntryPoint: + def load(self): + @cli.jobm.job("test_importlib38") + def f(): + app.config["IMPORTLIB38_TEST"] = "success" + return f + + # Mock Python 3.8-3.9 API: entry_points() returns dict-like object + class MockEntryPointsDict: + def __init__(self): + self._data = { + "sea.jobs": [EntryPoint()] + } + + def get(self, key, default=None): + return self._data.get(key, default) + + def mock_entry_points(group=None): + if group is not None: + # Python 3.10+ style call - should raise TypeError in 3.8-3.9 + raise TypeError("entry_points() takes 0 positional arguments") + return MockEntryPointsDict() + + mock_logger = mock.Mock() + with mock.patch("sea.cli.entry_points", side_effect=mock_entry_points), \ + mock.patch("logging.getLogger", return_value=mock_logger), \ + mock.patch("os.getcwd", return_value=app.root_path): + # Reload jobs to pick up the mocked entry points + cli._load_jobs() + sys.argv = "sea test_importlib38".split() + assert cli.main() is None + assert app.config.get("IMPORTLIB38_TEST") == "success" + + +def test_cmd_async_task_and_bus_registered(): + """Test that async_task and bus commands are registered via entry points""" + # This test verifies that the real entry points from setup.py are loaded + # Check that async_task and bus are registered after _load_jobs() + # Note: This will only work if sea package is installed with entry points + assert "async_task" in cli.jobm.jobs or "bus" in cli.jobm.jobs, \ + "async_task or bus should be registered via entry points. " \ + "Make sure sea package is installed with 'pip install -e .'" + + # If they exist, verify they are the correct functions + if "async_task" in cli.jobm.jobs: + assert cli.jobm.jobs["async_task"].__name__ == "async_task" + if "bus" in cli.jobm.jobs: + assert cli.jobm.jobs["bus"].__name__ == "bus" + + def test_main(): sys.argv = "sea -h".split() with pytest.raises(SystemExit): diff --git a/tests/test_server.py b/tests/test_server.py index f0d7761..f295f7c 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -36,42 +36,3 @@ def _mocked(*args, **kwargs): content = logstream.getvalue() assert "started!" in content and "stopped!" in content - - -def test_multiprocessing_server(app, logstream): - # In multiprocessing mode, prometheus dir must be set - try: - os.mkdir("/tmp/prometheus_metrics") - os.environ.setdefault("PROMETHEUS_MULTIPROC_DIR", "/tmp/prometheus_metrics") - - app.config["PROMETHEUS_PORT"] = 9092 - - from sea.server.multiprocessing import Server - - s = Server(app) - assert not s._stopped - - def log_started(s): - app.logger.warning("started!") - - def log_stopped(s): - app.logger.warning("stopped!") - - server_started.connect(log_started) - server_stopped.connect(log_stopped) - - def kill_later(sec): - time.sleep(sec) - os.kill(os.getpid(), signal.SIGINT) - - # 3 seconds to wait before killing server - threading.Thread(target=kill_later, args=[3]).start() - - # with mock.patch("time.sleep", new=_mocked): - assert s.run() - assert s._stopped - - content = logstream.getvalue() - assert "stopped!" in content - finally: - os.rmdir("/tmp/prometheus_metrics")