init commit

Author: robswc

requirements.txt

@@ -0,0 +1,7 @@
+click==8.0.4
+Flask==2.0.3
+itsdangerous==2.1.2
+Jinja2==3.1.1
+MarkupSafe==2.1.1
+typer==0.4.0
+Werkzeug==2.0.3

src/components/actions/base/action.py

@@ -0,0 +1,61 @@
+import datetime
+from logging import getLogger, DEBUG
+
+from utils.log import CONSOLE_HANDLER
+
+logger = getLogger(__name__)
+logger.addHandler(CONSOLE_HANDLER)
+logger.setLevel(DEBUG)
+
+
+class ActionManager:
+    def __init__(self):
+        self._actions = []
+
+    def get(self, action_name: str):
+        """
+        Gets action from manager that matches given name
+        :param action_name: name of action
+        :return: Action()
+        """
+        for action in self._actions:
+            if action.name == action_name:
+                return action
+
+        raise ValueError(f'Cannot find action with name {action_name}')
+
+
+am = ActionManager()
+
+
+class ActionLogEvent:
+    def __init__(self, status, msg):
+        self.timestamp = datetime.datetime.now()
+        self.status = status
+        self.msg = msg
+
+
+class Action:
+    objects = am
+
+    def __init__(self):
+        self.name = 'Unnamed Action'
+        self.logs = []
+
+    def __str__(self):
+        return f'{self.name}'
+
+    def get_logs(self):
+        """
+        Gets run logs in descending order
+        :return: list
+        """
+        return self.logs
+
+    def run(self, *args, **kwargs):
+        """
+        Runs, logs action
+        """
+        self.logs.append(ActionLogEvent('INFO', 'action run'))
+        logger.info(f'ACTION TRIGGERED --->\t{str(self)}')
+        pass

src/components/actions/base/template/action_template.py

@@ -0,0 +1,17 @@
+from components.actions.base.action import Action
+
+
+class TemplateActionClass(Action):
+    def __init__(self):
+        super().__init__()
+        self.name = '_TemplateAction_'
+
+    def run(self, *args, **kwargs):
+        super(Action).run(*args, **kwargs)
+        """
+        Custom run method. Add your custom logic here.
+        """
+        print(self.name, '---> action has run!')
+
+
+template_action = TemplateActionClass()

src/components/events/base/event.py

@@ -0,0 +1,30 @@
+# configure logging
+from logging import getLogger, DEBUG
+
+from utils.log import CONSOLE_HANDLER
+
+logger = getLogger(__name__)
+logger.addHandler(CONSOLE_HANDLER)
+logger.setLevel(DEBUG)
+
+
+class Event:
+    def __init__(self):
+        self.name = 'Unnamed Event'
+        self._actions = []
+        self.logs = []
+
+    def __str__(self):
+        return f'{self.name}'
+
+    def register_action(self, action):
+        """
+        Will implement checking here eventually (tm)
+        :param action: Action() to register
+        """
+        self._actions.append(action)
+
+    def trigger_actions(self):
+        logger.info(f'EVENT TRIGGERED --->\t{str(self)}')
+        for action in self._actions:
+            action.run()

src/components/events/base/template/event_template.py

@@ -0,0 +1,10 @@
+from components.events.base.event import Event
+
+
+class TemplateEventClass(Event):
+    def __init__(self):
+        super().__init__()
+        self.name = '_TemplateEvent_'
+
+
+template_event = TemplateEventClass()

src/components/schemas/order.py

@@ -0,0 +1,11 @@
+from components.schemas.schema import Schema
+
+
+class Order(Schema):
+    def __init__(self):
+        super().__init__()
+        self.order_type = str
+        self.side = str
+        self.quantity = int
+        self.symbol = str
+        self.price = float

src/components/schemas/schema.py

@@ -0,0 +1,7 @@
+
+class Schema:
+    def __init__(self):
+        pass
+
+    def as_json(self):
+        pass

src/main.py

@@ -0,0 +1,29 @@
+# initialize our Flask application
+from logging import getLogger, DEBUG
+
+from flask import Flask, request, jsonify, render_template
+
+from utils.log import CONSOLE_HANDLER
+
+app = Flask(__name__)
+
+# configure logging
+logger = getLogger(__name__)
+logger.addHandler(CONSOLE_HANDLER)
+logger.setLevel(DEBUG)
+
+
+@app.route("/", methods=["GET"])
+def dashboard():
+    if request.method == 'GET':
+        return render_template('dashboard.html')
+
+@app.route("/webhook", methods=["POST"])
+def webhook():
+    if request.method == 'POST':
+        logger.info(f'Request Data: {request.get_json()}')
+        return jsonify(str('success'))
+
+
+if __name__ == '__main__':
+    app.run(debug=True)

src/tests/__init__.py

@@ -0,0 +1,28 @@
+import os
+from unittest import TestCase
+
+
+class TestCLI(TestCase):
+    # change dir before running tests
+
+    os.chdir('..')
+
+    def test_newevent(self):
+        from tvwb import newevent
+        assert newevent(name='TestEvent')
+        self.assertRaises(ValueError, newevent, name='!')
+        self.assertRaises(ValueError, newevent, name='@')
+        self.assertRaises(ValueError, newevent, name='#')
+        self.assertRaises(ValueError, newevent, name='test event')
+        self.assertRaises(ValueError, newevent, name='test-event')
+        self.assertRaises(ValueError, newevent, name='test_event')
+
+    def test_newaction(self):
+        from tvwb import newaction
+        assert newaction(name='TestAction')
+        self.assertRaises(ValueError, newaction, name='!')
+        self.assertRaises(ValueError, newaction, name='@')
+        self.assertRaises(ValueError, newaction, name='#')
+        self.assertRaises(ValueError, newaction, name='test action')
+        self.assertRaises(ValueError, newaction, name='test-action')
+        self.assertRaises(ValueError, newaction, name='test_action')