Merge pull request #10

feat-web-socket

Author: dbunt1tled

.env.example

@@ -8,12 +8,11 @@ APP_NAME="App"
 SERVICE_NAME="api"
 X_API_KEY=
 
-# CORS Settings
-CORS_ALLOW_ORIGINS=["*"]
+CORS_ALLOW_ORIGINS='["*"]'
 CORS_ALLOW_CREDENTIALS=True
-CORS_ALLOW_METHODS=["*"]
-CORS_ALLOW_HEADERS=["*"]
-CORS_EXPOSE_HEADERS=["X-Total-Count", "X-Per-Page", "X-Current-Page", "X-Total-Pages", "X-User-Role"]
+CORS_ALLOW_METHODS='["*"]'
+CORS_ALLOW_HEADERS='["*"]'
+CORS_EXPOSE_HEADERS='["X-Total-Count", "X-Per-Page", "X-Current-Page", "X-Total-Pages", "X-User-Role"]'
 
 SQLALCHEMY_DATABASE_URI=mysql+aiomysql://user:password@host/db_name
 

pyproject.toml

@@ -23,6 +23,7 @@ dependencies = [
     "python-logging-loki>=0.3.1",
     "sqlalchemy>=2.0.41",
     "uvicorn-worker>=0.3.0",
+    "uvicorn[standard]>=0.35.0",
     "uvloop>=0.21.0",
 ]
 

server.py

@@ -29,7 +29,7 @@ def load(self) -> BaseApplication:
 if __name__ == "__main__":
     options = {
         "bind": "0.0.0.0:8000",
-        "workers": 1, # multiprocessing.cpu_count() * 2 + 1,
-        "worker_class": "uvicorn.workers.UvicornWorker",
+        "workers": 3, # multiprocessing.cpu_count() * 2 + 1,
+        "worker_class": "uvicorn_worker.UvicornWorker",
     }
     StandaloneApplication(application=app, option=options).run()

src/api.py

@@ -10,6 +10,7 @@
 from src.app.auth.controller.auth_controller import AuthController
 from src.app.user.controller.user_controller import UserController
 from src.app.user_notification.controller.user_notification_controller import UserNotificationController
+from src.app.ws.route.ws_controller import WSController
 from src.core.di.container import Container
 from src.core.enum.env import Env
 from src.core.exception.error_no import ErrorNo
@@ -31,8 +32,22 @@ async def lifespan(api: FastAPI) -> AsyncGenerator[None]:
     AuthController(app=api, container=container)
     UserController(app=api, container=container)
     UserNotificationController(app=api, container=container)
+    WSController(app=api, container=container)
+
+    # for route in api.routes:
+    #     if hasattr(route, "methods"):  # HTTP
+    #         methods = ",".join(route.methods)
+    #         print(f"HTTP     {methods:<10} {route.path}")
+    #     elif route.__class__.__name__ == "APIWebSocketRoute":
+    #         print(f"WS       {'-':<10} {route.path}")
+    #     else:
+    #         print(f"UNKNOWN  {'-':<10} {route.path}")
+
     yield
     await container.db_config().close()
+    await container.rmq_producer().close()
+    await container.rmq_consumer().close()
+    await container.ws_manager().close_all()
     container.unwire()
 
 
@@ -78,7 +93,7 @@ async def exception_handler(request: Request, e: Exception) -> JSONResponse:
         raise ValueError(e.args)
 
     if isinstance(e, RequestValidationError):
-        error = ApiResponseService.format_pydantic_error(errors=e.errors(), env=di.app_config().environment)
+        error = ApiResponseService.format_pydantic_error(errors=e.errors())
         di.log().error(message=f"RequestValidationError: {e}", error=str(error), request=req)
     elif isinstance(e, DomainException):
         e_data = e.as_dict()

src/app/auth/controller/auth_controller.py

@@ -5,8 +5,7 @@
 from src.app.auth.dto.login import LoginRequest
 from src.app.auth.dto.re_send_confirm_email import ReSendConfirmEmailRequest
 from src.app.auth.dto.sign_up import SignupRequest
-from src.cmd.worker.email.email_action import EmailAction
-from src.core.db.repository import Filter, FilterOperator
+from src.core.db.repository import Filter, Oper
 from src.core.di.container import Container
 from src.core.dto.dto import Message
 from src.core.exception.error_no import ErrorNo
@@ -42,7 +41,7 @@ async def re_send_confirm_email(self, req: ReSendConfirmEmailRequest) -> JsonApi
         res = Message(message="Email successfully sent")
         user = await self.container.user_service().one(
             filters=[
-                Filter("email", FilterOperator.EQ, req.email),
+                Filter("email", Oper.EQ, req.email),
             ]
         )
         if user is None:

src/app/auth/service/auth_service.py

@@ -2,7 +2,7 @@
 from src.app.user.data.user_status import UserStatus
 from src.app.user.model.user import User
 from src.app.user.service.user_service import UserService
-from src.core.db.repository import Filter, FilterOperator
+from src.core.db.repository import Filter, Oper
 from src.core.exception.error_no import ErrorNo
 from src.core.exception.exceptions import UnauthorizedException, UnprocessableEntityException
 from src.core.service.dto.token import TokenBearer, TokenType
@@ -24,7 +24,7 @@ def __init__(
     async def login(self, email: str, password: str) -> TokenBearer:
         user = await self.user_service.one(
             filters=[
-                Filter("email", FilterOperator.EQ, email),
+                Filter("email", Oper.EQ, email),
             ]
         )
         if user is None:
@@ -34,7 +34,7 @@ async def login(self, email: str, password: str) -> TokenBearer:
 
         if not self.hash_service.verify_password(password=password, hashed_password=user.hash_password):
             raise UnauthorizedException(error_no=ErrorNo.AUTHORIZATION_USER_PASSWORD_INVALID, message="Unauthorized!")
-        user = await self.user_service.update(id_value=user.id, data={"session": self.hash_service.random_string()})
+        user = await self.user_service.update(uid=user.id, data={"session": self.hash_service.random_string()})
 
         return self.hash_service.create_token_bearer(user=user)
 
@@ -47,7 +47,7 @@ async def signup(
     ) -> User:
         user = await self.user_service.one(
             filters=[
-                Filter("email", FilterOperator.EQ, email),
+                Filter("email", Oper.EQ, email),
             ]
         )
         if user is not None:
@@ -88,7 +88,7 @@ async def confirm_user(self, jwt: str) -> None:
             )
         user = await self.user_service.one(
             filters=[
-                Filter("email", FilterOperator.EQ, token.email),
+                Filter("email", Oper.EQ, token.email),
             ]
         )
         if user is None:
@@ -99,7 +99,7 @@ async def confirm_user(self, jwt: str) -> None:
             raise UnprocessableEntityException(
                 error_no=ErrorNo.CONFIRM_TOKEN_USER_SESSION_INVALID, message="Confirmation token is invalid"
             )
-        await self.user_service.update(id_value=user.id, data={"status": UserStatus.ACTIVE})
+        await self.user_service.update(uid=user.id, data={"status": UserStatus.ACTIVE})
 
     async def refresh(self, jwt: str) -> TokenBearer:
         token = self.hash_service.verify_token(token=jwt)
@@ -108,7 +108,7 @@ async def refresh(self, jwt: str) -> TokenBearer:
         if token.token_type != TokenType.REFRESH:
             raise UnauthorizedException(error_no=ErrorNo.REFRESH_TOKEN_TYPE_INVALID, message="Unauthorized!")
 
-        user = await self.user_service.get_by_id(id_value=int(token.subject))
+        user = await self.user_service.get_by_id(uid=int(token.subject))
         if user is None:
             raise UnauthorizedException(error_no=ErrorNo.REFRESH_TOKEN_USER_NOT_FOUND, message="Unauthorized!")
         if user.status != UserStatus.ACTIVE:
@@ -117,6 +117,6 @@ async def refresh(self, jwt: str) -> TokenBearer:
         if user.session != token.session:
             raise UnauthorizedException(error_no=ErrorNo.REFRESH_TOKEN_USER_SESSION_INVALID, message="Unauthorized!")
 
-        user = await self.user_service.update(id_value=user.id, data={"session": self.hash_service.random_string()})
+        user = await self.user_service.update(uid=user.id, data={"session": self.hash_service.random_string()})
 
         return self.hash_service.create_token_bearer(user=user)

src/app/user/controller/user_controller.py

@@ -1,7 +1,7 @@
 from fastapi import APIRouter, Depends, FastAPI, Request
 
 from src.app.user.dto.user import UserCreateRequest, UserListRequest
-from src.core.db.repository import Filter, FilterOperator, Pagination
+from src.core.db.repository import Filter, Oper, Pagination
 from src.core.di.container import Container
 from src.core.exception.error_no import ErrorNo
 from src.core.exception.exceptions import UnprocessableEntityException
@@ -21,7 +21,7 @@ def __init__(self, app: FastAPI, container: Container) -> None:
     async def list(self, req: UserListRequest = Depends()) -> JsonApiResponse:
         users = await self.container.user_service().all(
             filters=[
-                Filter("email", FilterOperator.EQ, req.email),
+                Filter("email", Oper.EQ, req.email),
             ],
             pagination=Pagination(
                 per_page=req.per_page or 10,
@@ -41,7 +41,7 @@ async def create(
     ) -> JsonApiResponse:
         user = await self.container.user_service().one(
             filters=[
-                Filter("email", FilterOperator.EQ, req.email),
+                Filter("email", Oper.EQ, req.email),
             ]
         )
         if user is not None:

src/app/user/service/user_service.py

@@ -10,17 +10,17 @@ class UserService:
     def __init__(self, user_repository: UserRepository) -> None:
         self.user_repository = user_repository
 
-    async def get_by_id(self, id_value: int) -> User:
-        return await self.user_repository.get_by_id(id_value=id_value)
+    async def get_by_id(self, uid: int) -> User:
+        return await self.user_repository.get_by_id(uid=uid)
 
-    async def find_by_id(self, id_value: int) -> User | None:
-        return await self.user_repository.find_by_id(id_value=id_value)
+    async def find_by_id(self, uid: int) -> User | None:
+        return await self.user_repository.find_by_id(uid=uid)
 
     async def create(self, data: dict[str, Any] | User) -> User:
         return await self.user_repository.create(data=data)
 
-    async def update(self, id_value: int, data: dict[str, Any]) -> User:
-        return await self.user_repository.update(id_value=id_value, data=data)
+    async def update(self, uid: int, data: dict[str, Any]) -> User:
+        return await self.user_repository.update(uid=uid, data=data)
 
     async def one(
         self,

src/app/user_notification/controller/user_notification_controller.py

@@ -2,7 +2,7 @@
 
 from src.app.user_notification.data.user_notification_status import UserNotificationStatus
 from src.app.user_notification.dto.user_notification import UserNotificationCreateRequest, UserNotificationListRequest
-from src.core.db.repository import Filter, FilterOperator, Pagination
+from src.core.db.repository import Filter, Oper, Pagination
 from src.core.di.container import Container
 from src.core.http.controller import BaseController
 from src.core.http.request.state import AuthState, get_auth_state
@@ -18,12 +18,14 @@ def __init__(self, app: FastAPI, container: Container) -> None:
         app.include_router(router=router)
 
     async def user_list(
-        self, state: AuthState = Depends(get_auth_state), req: UserNotificationListRequest = Depends()
+        self,
+        state: AuthState = Depends(get_auth_state),
+        req: UserNotificationListRequest = Depends(),
     ) -> JsonApiResponse:
         notifications = await self.container.user_notification_service().all(
             filters=[
-                Filter("user_id", FilterOperator.EQ, state.user.id),
-                Filter("status", FilterOperator.EQ, req.status),
+                Filter("user_id", Oper.EQ, state.user.id),
+                Filter("status", Oper.EQ, req.status),
             ],
             pagination=Pagination(
                 per_page=req.per_page or 10,

src/app/user_notification/model/user_notification.py

@@ -1,4 +1,6 @@
-from sqlalchemy import JSON, DateTime, Integer, String, ForeignKey
+from datetime import datetime
+
+from sqlalchemy import JSON, DateTime, ForeignKey, Integer
 from sqlalchemy.orm import Mapped, mapped_column
 from sqlalchemy.sql import func
 
@@ -18,11 +20,13 @@ class UserNotification(Entity):
     user_id: Mapped[int] = mapped_column(Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False)
     data: Mapped[dict] = mapped_column(JSON, nullable=False, default=dict)
     status: Mapped[UserNotificationStatus] = mapped_column(
-        IntEnum(UserNotificationStatus), nullable=False, default=UserStatus.PENDING, index=True
+        IntEnum(UserNotificationStatus), nullable=False, default=UserNotificationStatus.NEW, index=True
     )
-    updated_at: Mapped[DateTime] = mapped_column(
+    updated_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True),
         server_default=func.now(),
         onupdate=func.now(),
     )
-    created_at: Mapped[DateTime] = mapped_column(DateTime(timezone=True), server_default=func.now(), index=True)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), index=True
+    )