API scoped requests #186

example/custom/api/src/main.py

@@ -6,8 +6,10 @@ from cpl.application import ApplicationBuilder
 from cpl.auth.permission.permissions import Permissions
 from cpl.auth.schema import AuthUser, Role
 from cpl.core.configuration import Configuration
+from cpl.core.console import Console
 from cpl.core.environment import Environment
 from cpl.core.utils.cache import Cache
+from custom.api.src.scoped_service import ScopedService
 from service import PingService
 
 
@@ -23,6 +25,8 @@ def main():
     builder.services.add_transient(PingService)
     builder.services.add_module(api)
 
+    builder.services.add_scoped(ScopedService)
+
     builder.services.add_cache(AuthUser)
     builder.services.add_cache(Role)
 
@@ -40,6 +44,32 @@ def main():
     user_cache = provider.get_service(Cache[AuthUser])
     role_cache = provider.get_service(Cache[Role])
 
+    if role_cache == user_cache:
+        raise Exception("Cache service is not working")
+
+    s1 = provider.get_service(ScopedService)
+    s2 = provider.get_service(ScopedService)
+
+    if s1.name == s2.name:
+        raise Exception("Scoped service is not working")
+
+    with provider.create_scope() as scope:
+        s3 = scope.get_service(ScopedService)
+        s4 = scope.get_service(ScopedService)
+
+        if s3.name != s4.name:
+            raise Exception("Scoped service is not working")
+
+    if s1.name == s3.name:
+        raise Exception("Scoped service is not working")
+
+    Console.write_line(
+        s1.name,
+        s2.name,
+        s3.name,
+        s4.name,
+    )
+
     app.run()
 
 

example/custom/api/src/routes/ping.py

@@ -5,12 +5,17 @@ from starlette.responses import JSONResponse
 
 from cpl.api import APILogger
 from cpl.api.router import Router
+from cpl.core.console import Console
+from cpl.dependency import ServiceProvider
+from custom.api.src.scoped_service import ScopedService
 
 
 @Router.authenticate()
 # @Router.authorize(permissions=[Permissions.administrator])
 # @Router.authorize(policies=["test"])
 @Router.get(f"/ping")
-async def ping(r: Request, ping: PingService, logger: APILogger):
+async def ping(r: Request, ping: PingService, logger: APILogger, provider: ServiceProvider, scoped: ScopedService):
     logger.info(f"Ping: {ping}")
+
+    Console.write_line(scoped.name)
     return JSONResponse(ping.ping(r))

example/custom/api/src/scoped_service.py

@@ -0,0 +1,14 @@
+from cpl.core.console.console import Console
+from cpl.core.utils.string import String
+
+
+class ScopedService:
+    def __init__(self):
+        self._name = String.random(8)
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    def run(self):
+        Console.write_line(f"Im {self._name}")