💥 NEW: plugin support (close #2)

examples/asgi/app.py

@@ -2,27 +2,36 @@
 
 from falcon.asgi.ws import WebSocket
 
-from falca.responses import HTMLResponse
+from falca.responses import HTMLResponse, JSONResponse
 
 envvar = "FALCA_SETTINGS"
 os.environ.setdefault(envvar, "settings")
 
+from plugin import Logger
 from private.router import private_router
 from resources.article import Article
 from resources.home import Home
 from resources.media import Media
 
 from falca.app import ASGI
+from falca.depends import Plugin, Settings
 
 app = ASGI(__name__)
 app.settings.from_envvar(envvar)
+app.plugins.install("plugin.Logger")
 
 
 @app.get("/")
-async def index():
+async def index(logger: Logger = Plugin("logging")):
+    logger.info("are you ok ?")
     return HTMLResponse("index.html", context={"body": "not bad!"})
 
 
+@app.get("/settings")
+async def settings(part_config: dict = Settings("part_config")):
+    return JSONResponse(part_config)
+
+
 @app.websocket("/events")
 async def events(ws: WebSocket):
     await ws.accept()

examples/asgi/plugin.py

@@ -0,0 +1,29 @@
+import logging
+from typing import Union
+
+from falca.app import ASGI, WSGI
+from falca.plugins.base import BasePlugin
+
+
+class Logger(BasePlugin):
+    name = "logging"
+
+    def __init__(self, app: Union[WSGI, ASGI]) -> None:
+        super().__init__(app)
+        logging.basicConfig(level=logging.INFO)
+        self._log = logging.getLogger(app.import_name)
+
+    def info(self, *args, **kwds):
+        self._log.info(*args, **kwds)
+
+    def error(self, *args, **kwds):
+        self._log.error(*args, **kwds)
+
+    def warning(self, *args, **kwds):
+        self._log.warning(*args, **kwds)
+
+    def exception(self, *args, **kwds):
+        self._log.exception(*args, **kwds)
+
+    def debug(self, *args, **kwds):
+        self._log.debug(*args, **kwds)

examples/wsgi/app.py

@@ -3,23 +3,32 @@
 envvar = "FALCA_SETTINGS"
 os.environ.setdefault(envvar, "settings")
 
+from plugin import Logger
 from private.router import private_router
 from resources.article import Article
 from resources.home import Home
 from resources.media import Media
 
 from falca.app import WSGI
-from falca.responses import HTMLResponse
+from falca.depends import Plugin, Settings
+from falca.responses import HTMLResponse, JSONResponse
 
 app = WSGI(__name__)
 app.settings.from_envvar(envvar)
+app.plugins.install("plugin.Logger")
 
 
 @app.get("/")
-def index():
+def index(logger: Logger = Plugin("logging")):
+    logger.info("are you ok ?")
     return HTMLResponse("index.html", context={"body": "not bad!"})
 
 
+@app.get("/settings")
+def settings(part_config: dict = Settings("part_config")):
+    return JSONResponse(part_config)
+
+
 app.add_route("/home", Home())
 app.add_route("/article", Article())
 app.add_route("/form", Article(), suffix="form")

examples/wsgi/plugin.py

@@ -0,0 +1,29 @@
+import logging
+from typing import Union
+
+from falca.app import ASGI, WSGI
+from falca.plugins.base import BasePlugin
+
+
+class Logger(BasePlugin):
+    name = "logging"
+
+    def __init__(self, app: Union[WSGI, ASGI]) -> None:
+        super().__init__(app)
+        logging.basicConfig(level=logging.INFO)
+        self._log = logging.getLogger(app.import_name)
+
+    def info(self, *args, **kwds):
+        self._log.info(*args, **kwds)
+
+    def error(self, *args, **kwds):
+        self._log.error(*args, **kwds)
+
+    def warning(self, *args, **kwds):
+        self._log.warning(*args, **kwds)
+
+    def exception(self, *args, **kwds):
+        self._log.exception(*args, **kwds)
+
+    def debug(self, *args, **kwds):
+        self._log.debug(*args, **kwds)

falca/exceptions.py

@@ -2,7 +2,7 @@ class FalcaError(Exception):
     pass
 
 
-class PluginNotFound(FalcaError):
+class PluginError(FalcaError):
     pass
 
 

falca/plugins/base.py

@@ -0,0 +1,5 @@
+class BasePlugin:
+    name: str = None
+
+    def __init__(self, app) -> None:
+        self.app = app

falca/plugins/manager.py

@@ -2,22 +2,33 @@
 
 from falcon.app import App
 
+from ..exceptions import PluginError
 from ..helpers import import_attr
+from .base import BasePlugin
 
 
 class PluginManager:
     def __init__(self, app: App) -> None:
         self.app = app
-        self.storage: Dict[str, object] = {}
+        self.storage: Dict[str, BasePlugin] = {}
 
     def has(self, name: str):
         return name in self.storage
 
     def get(self, name: str):
+        if not self.has(name):
+            raise PluginError(f"plugin {name!r} not found")
         return self.storage.get(name)
 
-    def install(self, name: str, src: str):
+    def install(self, src: str, name: str = None):
         plugin = import_attr(src)
+        if not issubclass(plugin, BasePlugin):
+            raise PluginError(f"invalid plugin object {plugin}")
+
+        name = name or plugin.name
+        if not name:
+            raise PluginError("plugin name is required")
+
         self.storage[name] = plugin(self.app)
 
     def uninstall(self, name: str):