platypush/platypush/__init__.py

283 lines
8.5 KiB
Python
Raw Normal View History

2018-07-30 22:08:06 +02:00
"""
Platypush
.. moduleauthor:: Fabio Manganiello <blacklight86@gmail.com>
2018-07-30 23:18:01 +02:00
.. license: MIT
2018-07-30 22:08:06 +02:00
"""
import argparse
2017-10-29 05:17:42 +01:00
import logging
2018-12-19 21:15:06 +01:00
import os
import sys
2018-09-20 12:49:57 +02:00
from .bus.redis import RedisBus
2017-12-18 01:10:51 +01:00
from .config import Config
2021-07-22 01:02:15 +02:00
from .context import register_backends, register_plugins
2018-01-15 22:36:24 +01:00
from .cron.scheduler import CronScheduler
from .event.processor import EventProcessor
from .logger import Logger
from .message.event import Event
from .message.event.application import ApplicationStartedEvent
from .message.request import Request
from .message.response import Response
2021-07-22 01:02:15 +02:00
from .utils import set_thread_name, get_enabled_plugins
2017-10-29 05:17:42 +01:00
2020-09-27 01:33:38 +02:00
__author__ = 'Fabio Manganiello <info@fabiomanganiello.com>'
2022-12-10 15:37:05 +01:00
__version__ = '0.24.2'
2017-11-03 01:56:27 +01:00
2020-09-27 01:33:38 +02:00
logger = logging.getLogger('platypush')
2018-06-06 20:09:18 +02:00
2018-07-30 22:08:06 +02:00
class Daemon:
"""Main class for the Platypush daemon"""
2018-07-30 22:08:06 +02:00
# Configuration file (default: either ~/.config/platypush/config.yaml or
# /etc/platypush/config.yaml
config_file = None
2018-07-30 22:08:06 +02:00
# Application bus. It's an internal queue where:
# - backends will post the messages they receive
# - plugins will post the responses they process
bus = None
# Default bus queue name
_default_redis_queue = 'platypush/bus'
2018-12-19 21:15:06 +01:00
pidfile = None
2018-07-30 22:08:06 +02:00
# backend_name => backend_obj map
backends = None
2018-07-30 22:08:06 +02:00
# number of executions retries before a request fails
n_tries = 2
def __init__(
self,
config_file=None,
pidfile=None,
requests_to_process=None,
no_capture_stdout=False,
no_capture_stderr=False,
redis_queue=None,
verbose=False,
):
2018-07-30 22:08:06 +02:00
"""
Constructor
Params:
config_file -- Configuration file override (default: None)
2018-12-19 21:15:06 +01:00
pidfile -- File where platypush will store its PID upon launch,
useful if you're planning to integrate the application
within a service or a launcher script (default: None)
2017-12-22 02:11:56 +01:00
requests_to_process -- Exit after processing the specified number
of requests (default: None, loop forever)
no_capture_stdout -- Set to true if you want to disable the stdout
capture by the logging system
no_capture_stderr -- Set to true if you want to disable the stderr
capture by the logging system
redis_queue -- Name of the (Redis) queue used for dispatching messages (default: platypush/bus).
verbose -- Enable debug/verbose logging, overriding the stored configuration (default: False).
"""
2018-12-19 21:15:06 +01:00
if pidfile:
self.pidfile = pidfile
with open(self.pidfile, 'w') as f:
2018-12-19 21:24:39 +01:00
f.write(str(os.getpid()))
2018-12-19 21:15:06 +01:00
self.redis_queue = redis_queue or self._default_redis_queue
self.config_file = config_file
Config.init(self.config_file)
logging_conf = Config.get('logging') or {}
if verbose:
logging_conf['level'] = logging.DEBUG
logging.basicConfig(**logging_conf)
redis_conf = Config.get('backend.redis') or {}
self.bus = RedisBus(
redis_queue=self.redis_queue,
on_message=self.on_message(),
**redis_conf.get('redis_args', {})
)
self.no_capture_stdout = no_capture_stdout
self.no_capture_stderr = no_capture_stderr
self.event_processor = EventProcessor()
2017-12-22 02:11:56 +01:00
self.requests_to_process = requests_to_process
self.processed_requests = 0
self.cron_scheduler = None
2017-12-22 02:11:56 +01:00
@classmethod
def build_from_cmdline(cls, args):
2018-07-30 22:08:06 +02:00
"""
Build the app from command line arguments.
Params:
args -- Your sys.argv[1:] [List of strings]
"""
parser = argparse.ArgumentParser()
parser.add_argument(
'--config',
'-c',
dest='config',
required=False,
default=None,
help=cls.config_file.__doc__,
)
parser.add_argument(
'--version',
dest='version',
required=False,
action='store_true',
help="Print the current version and exit",
)
parser.add_argument(
'--verbose',
'-v',
dest='verbose',
required=False,
action='store_true',
help="Enable verbose/debug logging",
)
parser.add_argument(
'--pidfile',
'-P',
dest='pidfile',
required=False,
default=None,
help="File where platypush will "
+ "store its PID, useful if you're planning to "
+ "integrate it in a service",
)
parser.add_argument(
'--no-capture-stdout',
dest='no_capture_stdout',
required=False,
action='store_true',
help="Set this flag if you have max stack depth "
+ "exceeded errors so stdout won't be captured by "
+ "the logging system",
)
parser.add_argument(
'--no-capture-stderr',
dest='no_capture_stderr',
required=False,
action='store_true',
help="Set this flag if you have max stack depth "
+ "exceeded errors so stderr won't be captured by "
+ "the logging system",
)
parser.add_argument(
'--redis-queue',
dest='redis_queue',
required=False,
default=cls._default_redis_queue,
help="Name of the Redis queue to be used to internally deliver messages "
"(default: platypush/bus)",
)
opts, args = parser.parse_known_args(args)
if opts.version:
print(__version__)
sys.exit(0)
return cls(
config_file=opts.config,
pidfile=opts.pidfile,
no_capture_stdout=opts.no_capture_stdout,
no_capture_stderr=opts.no_capture_stderr,
redis_queue=opts.redis_queue,
verbose=opts.verbose,
)
def on_message(self):
2018-07-30 22:08:06 +02:00
"""
Default message handler
"""
def _f(msg):
2018-07-30 22:08:06 +02:00
"""
on_message closure
Params:
2018-07-30 22:08:06 +02:00
msg -- platypush.message.Message instance
"""
if isinstance(msg, Request):
try:
msg.execute(n_tries=self.n_tries)
except PermissionError:
2020-09-27 01:33:38 +02:00
logger.info('Dropped unauthorized request: {}'.format(msg))
self.processed_requests += 1
if (
self.requests_to_process
and self.processed_requests >= self.requests_to_process
):
self.stop_app()
elif isinstance(msg, Response):
msg.log()
elif isinstance(msg, Event):
msg.log()
self.event_processor.process_event(msg)
return _f
2017-12-22 02:11:56 +01:00
def stop_app(self):
"""Stops the backends and the bus"""
2021-07-22 01:02:15 +02:00
from .plugins import RunnablePlugin
2017-12-22 02:11:56 +01:00
for backend in self.backends.values():
backend.stop()
2021-07-22 01:02:15 +02:00
for plugin in get_enabled_plugins().values():
if isinstance(plugin, RunnablePlugin):
plugin.stop()
2017-12-22 02:11:56 +01:00
self.bus.stop()
if self.cron_scheduler:
self.cron_scheduler.stop()
2017-12-22 02:11:56 +01:00
def run(self):
"""Start the daemon"""
if not self.no_capture_stdout:
2020-09-27 01:33:38 +02:00
sys.stdout = Logger(logger.info)
if not self.no_capture_stderr:
2020-09-27 01:33:38 +02:00
sys.stderr = Logger(logger.warning)
set_thread_name('platypush')
logger.info('---- Starting platypush v.{}'.format(__version__))
# Initialize the backends and link them to the bus
2017-12-24 13:15:37 +01:00
self.backends = register_backends(bus=self.bus, global_scope=True)
# Start the backend threads
for backend in self.backends.values():
backend.start()
2021-07-22 01:02:15 +02:00
# Initialize the plugins
register_plugins(bus=self.bus)
2018-01-15 22:36:24 +01:00
# Start the cron scheduler
if Config.get_cronjobs():
self.cron_scheduler = CronScheduler(jobs=Config.get_cronjobs())
self.cron_scheduler.start()
2018-01-15 22:36:24 +01:00
self.bus.post(ApplicationStartedEvent())
# Poll for messages on the bus
try:
self.bus.poll()
2018-07-30 22:08:06 +02:00
except KeyboardInterrupt:
2020-09-27 01:33:38 +02:00
logger.info('SIGINT received, terminating application')
2017-12-22 02:11:56 +01:00
finally:
self.stop_app()
2018-01-04 16:11:54 +01:00
def main():
2018-07-30 22:08:06 +02:00
"""
Platypush daemon main
"""
2018-01-04 16:11:54 +01:00
app = Daemon.build_from_cmdline(sys.argv[1:])
app.run()
2018-01-04 16:11:54 +01:00
# vim:sw=4:ts=4:et: