# pyfunceble-process-manager
**The process manager library for and from the PyFunceble project.**
This is a project component that was initially part of the PyFunceble project.
It aims to separate the process manager from the main project to allow others
to use it without the need to install the whole PyFunceble project.
# Installation
```sh
pip3 install pyfunceble-process-manager
```
## Reserved Messages
As we implement our very own "message system" to communicate between the
processes, we have reserved some messages that you should not use in your
implementation.
| Message | Description |
| ------------------------ | ----------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `__stop__` | This message is used to instruct the worker to start the shutdown process. |
| `__immediate_shutdown__` | This message is used to instruct the worker to shutdown immediately. |
| `__wait__` | This message is used to instruct the worker to wait for the next message. _Sending this message will actually make the worker re-execute any (preflight) checks._ |
## Example
```python
import logging
import sys
from typing import Any
from PyFunceble.ext.process_manager import ProcessManagerCore, WorkerCore
class DataFilterWorker(WorkerCore):
def perform_external_poweron_checks(self) -> bool:
# This can be used to perform checks before starting the worker.
# If False is returned, the worker will not start.
return super().perform_external_poweron_checks()
def perform_external_poweroff_checks(self) -> bool:
# This can be used to perform checks before stopping the worker.
return super().perform_external_poweroff_checks()
def perform_external_preflight_checks(self) -> bool:
# This can be used to implement your own logic before the worker
# starts processing the next data.
return super().perform_external_preflight_checks()
def perform_external_inflight_checks(self, consumed: Any) -> bool:
# This can be used to filter out the consumed data by returning False.
# For example, filter out data that is not a string.
print("DataFilter consumed:", consumed)
if not isinstance(consumed, str):
print("DataFilter filtered out:", consumed)
return False
return super().perform_external_inflight_checks(consumed)
def perform_external_postflight_checks(self, produced: Any) -> bool:
# This can be used to implement your own logic after the worker
# has processed the data.
print("DataFilter produced:", produced)
return super().perform_external_postflight_checks(produced)
def target(self, consumed: Any) -> Any:
# Here we simply convert the string to uppercase as an example of data
# processing.
return consumed.upper()
class DataPrinterWorker(WorkerCore):
def target(self, consumed: Any) -> Any:
print("DataPrinter consumed:", consumed)
return consumed
class DataFilterManager(ProcessManagerCore):
STD_NAME = "data-filter"
WORKER_CLASS = DataFilterWorker
class DataPrinterManager(ProcessManagerCore):
STD_NAME = "data-printer"
WORKER_CLASS = DataPrinterWorker
if __name__ == "__main__":
dynamic_scaling = len(sys.argv) > 1
# By default, our interfaces won't log anything. If you need to see or analyze
# what is going on under the hood, uncomment the following
# logging.basicConfig(level=logging.DEBUG)
# logging.getLogger("PyFunceble.ext.process_manager").setLevel(logging.DEBUG)
data_to_filter = [
"hello",
"world",
123, # This will be filtered out because it's not a string.
"PyFunceble",
None, # This will be filtered out because it's not a string.
]
if dynamic_scaling:
# Add more data so that we can see the workers in action.
data_to_filter += [f"Hello, {i}!" for i in range(1000 + 1)]
# Configure the manager to generate 2 workers/processes.
data_filter_manager = DataFilterManager(
max_workers=10,
generate_output_queue=True,
output_queue_count=1,
dynamic_up_scaling=dynamic_scaling,
dynamic_down_scaling=dynamic_scaling,
)
# Configure the manager to generate 1 worker/process.
data_printer_manager = DataPrinterManager(
max_workers=1,
input_queue=data_filter_manager.output_queues[0],
generate_output_queue=False,
)
if dynamic_scaling:
# Build dependencies, so that we can use scaling at it best.
data_filter_manager.add_dependent_manager(data_printer_manager)
# Start the manager.
data_filter_manager.start()
data_printer_manager.start()
# Push the data to the input queue for further processing.
for data in data_to_filter:
print("Controller pushed:", data)
data_filter_manager.push_to_input_queue(data)
# Push the stop signal at the end of the input queue to stop the workers.
# Note:
# As the data printer is a dependency of the data filter, we don't need to
# stop it explicitly. It will be stopped automatically when the data filter
# manager stops.
#
data_filter_manager.push_stop_signal()
# Wait for the manager to finish processing the data.
data_filter_manager.wait()
# If we want to terminate the manager and all workers, we can call the
# terminate method.
data_filter_manager.terminate()
print("Data filtered successfully.")
```
# License
Copyright 2017, 2018, 2019, 2020, 2022, 2023, 2024 Nissar Chababy
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.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.
Raw data
{
"_id": null,
"home_page": "https://github.com/PyFunceble/process-manager",
"name": "pyfunceble-process-manager",
"maintainer": null,
"docs_url": null,
"requires_python": "<4,>=3.8",
"maintainer_email": null,
"keywords": "PyFunceble, process-manager",
"author": "funilrys",
"author_email": "contact@funilrys.com",
"download_url": "https://files.pythonhosted.org/packages/a4/ec/f3f28cc783d98bc15c4f185fa82463e96c0a7a829b1f30d1666c46ced130/pyfunceble_process_manager-1.0.5.tar.gz",
"platform": "any",
"description": "\n\n# pyfunceble-process-manager\n\n**The process manager library for and from the PyFunceble project.**\n\nThis is a project component that was initially part of the PyFunceble project.\nIt aims to separate the process manager from the main project to allow others\nto use it without the need to install the whole PyFunceble project.\n\n# Installation\n\n```sh\npip3 install pyfunceble-process-manager\n```\n\n## Reserved Messages\n\nAs we implement our very own \"message system\" to communicate between the\nprocesses, we have reserved some messages that you should not use in your\nimplementation.\n\n| Message | Description |\n| ------------------------ | ----------------------------------------------------------------------------------------------------------------------------------------------------------------- |\n| `__stop__` | This message is used to instruct the worker to start the shutdown process. |\n| `__immediate_shutdown__` | This message is used to instruct the worker to shutdown immediately. |\n| `__wait__` | This message is used to instruct the worker to wait for the next message. _Sending this message will actually make the worker re-execute any (preflight) checks._ |\n\n\n## Example\n\n```python\nimport logging\nimport sys\nfrom typing import Any\n\nfrom PyFunceble.ext.process_manager import ProcessManagerCore, WorkerCore\n\n\nclass DataFilterWorker(WorkerCore):\n def perform_external_poweron_checks(self) -> bool:\n # This can be used to perform checks before starting the worker.\n # If False is returned, the worker will not start.\n return super().perform_external_poweron_checks()\n\n def perform_external_poweroff_checks(self) -> bool:\n # This can be used to perform checks before stopping the worker.\n return super().perform_external_poweroff_checks()\n\n def perform_external_preflight_checks(self) -> bool:\n # This can be used to implement your own logic before the worker\n # starts processing the next data.\n return super().perform_external_preflight_checks()\n\n def perform_external_inflight_checks(self, consumed: Any) -> bool:\n # This can be used to filter out the consumed data by returning False.\n # For example, filter out data that is not a string.\n\n print(\"DataFilter consumed:\", consumed)\n\n if not isinstance(consumed, str):\n\n print(\"DataFilter filtered out:\", consumed)\n\n return False\n return super().perform_external_inflight_checks(consumed)\n\n def perform_external_postflight_checks(self, produced: Any) -> bool:\n # This can be used to implement your own logic after the worker\n # has processed the data.\n\n print(\"DataFilter produced:\", produced)\n return super().perform_external_postflight_checks(produced)\n\n def target(self, consumed: Any) -> Any:\n # Here we simply convert the string to uppercase as an example of data\n # processing.\n return consumed.upper()\n\n\nclass DataPrinterWorker(WorkerCore):\n def target(self, consumed: Any) -> Any:\n print(\"DataPrinter consumed:\", consumed)\n return consumed\n\n\nclass DataFilterManager(ProcessManagerCore):\n STD_NAME = \"data-filter\"\n WORKER_CLASS = DataFilterWorker\n\n\nclass DataPrinterManager(ProcessManagerCore):\n STD_NAME = \"data-printer\"\n WORKER_CLASS = DataPrinterWorker\n\n\nif __name__ == \"__main__\":\n dynamic_scaling = len(sys.argv) > 1\n\n # By default, our interfaces won't log anything. If you need to see or analyze\n # what is going on under the hood, uncomment the following\n # logging.basicConfig(level=logging.DEBUG)\n # logging.getLogger(\"PyFunceble.ext.process_manager\").setLevel(logging.DEBUG)\n\n data_to_filter = [\n \"hello\",\n \"world\",\n 123, # This will be filtered out because it's not a string.\n \"PyFunceble\",\n None, # This will be filtered out because it's not a string.\n ]\n\n if dynamic_scaling:\n # Add more data so that we can see the workers in action.\n data_to_filter += [f\"Hello, {i}!\" for i in range(1000 + 1)]\n\n # Configure the manager to generate 2 workers/processes.\n data_filter_manager = DataFilterManager(\n max_workers=10,\n generate_output_queue=True,\n output_queue_count=1,\n dynamic_up_scaling=dynamic_scaling,\n dynamic_down_scaling=dynamic_scaling,\n )\n # Configure the manager to generate 1 worker/process.\n data_printer_manager = DataPrinterManager(\n max_workers=1,\n input_queue=data_filter_manager.output_queues[0],\n generate_output_queue=False,\n )\n\n if dynamic_scaling:\n # Build dependencies, so that we can use scaling at it best.\n data_filter_manager.add_dependent_manager(data_printer_manager)\n\n # Start the manager.\n data_filter_manager.start()\n data_printer_manager.start()\n\n # Push the data to the input queue for further processing.\n for data in data_to_filter:\n print(\"Controller pushed:\", data)\n data_filter_manager.push_to_input_queue(data)\n\n # Push the stop signal at the end of the input queue to stop the workers.\n # Note:\n # As the data printer is a dependency of the data filter, we don't need to\n # stop it explicitly. It will be stopped automatically when the data filter\n # manager stops.\n #\n data_filter_manager.push_stop_signal()\n # Wait for the manager to finish processing the data.\n data_filter_manager.wait()\n\n # If we want to terminate the manager and all workers, we can call the\n # terminate method.\n data_filter_manager.terminate()\n\n print(\"Data filtered successfully.\")\n\n```\n\n# License\n\n Copyright 2017, 2018, 2019, 2020, 2022, 2023, 2024 Nissar Chababy\n\n Licensed under the Apache License, Version 2.0 (the \"License\");\n you may not use this file except in compliance with the License.\n You may obtain a copy of the License at\n\n https://www.apache.org/licenses/LICENSE-2.0\n\n Unless required by applicable law or agreed to in writing, software\n distributed under the License is distributed on an \"AS IS\" BASIS,\n WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n See the License for the specific language governing permissions and\n limitations under the License.\n",
"bugtrack_url": null,
"license": "Apache 2.0",
"summary": "The process manager library for and from the PyFunceble project.",
"version": "1.0.5",
"project_urls": {
"Homepage": "https://github.com/PyFunceble/process-manager",
"Source": "https://github.com/PyFunceble/process-manager/tree/master",
"Tracker": "https://github.com/PyFunceble/process-manager/issues"
},
"split_keywords": [
"pyfunceble",
" process-manager"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "f27d88497cb7bb3378083998f6aa3a93101ae60ca853a0cfede159320c9ab8da",
"md5": "b157dd6a4c6ad3347e609d069530856e",
"sha256": "b7368280aecb911556198cc02f5bef9b64a28c8c0574a314fcd46acaa63a08a2"
},
"downloads": -1,
"filename": "pyfunceble_process_manager-1.0.5-py3-none-any.whl",
"has_sig": false,
"md5_digest": "b157dd6a4c6ad3347e609d069530856e",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4,>=3.8",
"size": 27064,
"upload_time": "2024-11-03T14:27:37",
"upload_time_iso_8601": "2024-11-03T14:27:37.608255Z",
"url": "https://files.pythonhosted.org/packages/f2/7d/88497cb7bb3378083998f6aa3a93101ae60ca853a0cfede159320c9ab8da/pyfunceble_process_manager-1.0.5-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "a4ecf3f28cc783d98bc15c4f185fa82463e96c0a7a829b1f30d1666c46ced130",
"md5": "926a2eebdf36d6700d8c7bcad21fb1f0",
"sha256": "69e6e40f637ad8075cb68519985e8900c824a4b55c7acff868d592a665e4b209"
},
"downloads": -1,
"filename": "pyfunceble_process_manager-1.0.5.tar.gz",
"has_sig": false,
"md5_digest": "926a2eebdf36d6700d8c7bcad21fb1f0",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4,>=3.8",
"size": 30286,
"upload_time": "2024-11-03T14:27:39",
"upload_time_iso_8601": "2024-11-03T14:27:39.623257Z",
"url": "https://files.pythonhosted.org/packages/a4/ec/f3f28cc783d98bc15c4f185fa82463e96c0a7a829b1f30d1666c46ced130/pyfunceble_process_manager-1.0.5.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-11-03 14:27:39",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "PyFunceble",
"github_project": "process-manager",
"travis_ci": false,
"coveralls": true,
"github_actions": true,
"requirements": [],
"tox": true,
"lcname": "pyfunceble-process-manager"
}
JSON
