monitor support and event support

mgship/cli.py

@@ -39,12 +39,29 @@ def to_timestamp(ctx, param, value):
 @click.option('--begin', default=None, type=DateTime(),
               callback=to_timestamp,
               help='when to start archiving, as unix timestamp')
+@click.option('--event', default=None, type=str,
+              help='event type')
 @click.option('--recipient', default=None, type=Email(),
               help='email address of recipient')
 def archive(format, *args, **kwargs):
     dest = csv.Destination() if format == 'csv' else json.Destination()
     mgship.Archive(dest, *args, **kwargs).ship()
 
 
+@main.command()
+@click.option('--format', default='csv', type=click.Choice(['json', 'csv']),
+              help='the output format')
+@click.option('--sleep', default=None, type=int,
+              help='how much to wait before repeating requests')
+@click.option('--begin', default=None, type=DateTime(),
+              callback=to_timestamp,
+              help='when to start polling, as unix timestamp')
+@click.option('--recipient', default=None, type=Email(),
+              help='email address of recipient')
+def monitor(format, *args, **kwargs):
+    dest = csv.Destination() if format == 'csv' else json.Destination()
+    mgship.Monitor(dest, *args, **kwargs).ship()
+
+
 if __name__ == "__main__":
     main()

mgship/events.py

@@ -50,6 +50,8 @@ def mg_poll_events(client=None, max_age=1800, sleep=900, rewind=None,
 
     if 'begin' not in kwargs:
         begin = kwargs['begin'] = utctimestamp(timeago(seconds=rewind))
+    else:
+        begin = kwargs['begin']
 
     url = events_url(ascending="yes", **kwargs)
 

mgship/mgship.py

@@ -6,7 +6,7 @@
 from functools import wraps
 
 from mgship.api import Client
-from mgship.events import mg_past_events
+from mgship.events import mg_past_events, mg_poll_events
 from mgship.util import is_past
 from mgship.email import is_email
 from mgship.log import logger
@@ -30,6 +30,7 @@ class Archive(object):
     """Ship all existing events."""
     dest = attr.ib()
     begin = attr.ib(default=None, validator=mg_field_validator(is_past))
+    event = attr.ib(default=None)
     recipient = attr.ib(default=None, validator=mg_field_validator(is_email))
     _client = attr.ib(default=attr.Factory(Client), repr=False)
     _filtered_params = ['dest', '_client']
@@ -55,3 +56,26 @@ def ship(self):
 @attr.s
 class Monitor(object):
     """Monitor current events."""
+    dest = attr.ib()
+    begin = attr.ib(default=None)
+    sleep = attr.ib(default=None)
+    recipient = attr.ib(default=None, validator=mg_field_validator(is_email))
+    _client = attr.ib(default=attr.Factory(Client), repr=False)
+    _filtered_params = ['dest', '_client']
+
+    @classmethod
+    def _filter_params(self, attr, value):
+        if attr.name in self._filtered_params or value is None:
+            return False
+        return True
+
+    def _iter_sink(self, sink):
+        kwargs = attr.asdict(self, filter=self._filter_params)
+        logger.info("starting monitoring ({})".format(kwargs))
+        for event in mg_poll_events(client=self._client, **kwargs):
+            yield sink.send(event)
+
+    def ship(self):
+        with self.dest as sink:
+            for _ in self._iter_sink(sink):
+                pass