feat-web-socket

.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
+    )

src/app/user_notification/service/user_notification_service.py

@@ -4,24 +4,24 @@
 from src.app.user_notification.data.user_notification_status import UserNotificationStatus
 from src.app.user_notification.model.user_notification import UserNotification
 from src.app.user_notification.repository.user_notification_repository import UserNotificationRepository
-from src.core.db.repository import Filter, FilterOperator, OrderBy, Pager, Pagination, Paginator
+from src.core.db.repository import Filter, Oper, OrderBy, Pager, Pagination, Paginator
 
 
 class UserNotificationService:
     def __init__(self, user_notification_repository: UserNotificationRepository) -> None:
         self.user_notification_repository = user_notification_repository
 
-    async def get_by_id(self, id_value: int) -> UserNotification:
-        return await self.user_notification_repository.get_by_id(id_value=id_value)
+    async def get_by_id(self, uid: int) -> UserNotification:
+        return await self.user_notification_repository.get_by_id(uid=uid)
 
-    async def find_by_id(self, id_value: int) -> UserNotification | None:
-        return await self.user_notification_repository.find_by_id(id_value=id_value)
+    async def find_by_id(self, uid: int) -> UserNotification | None:
+        return await self.user_notification_repository.find_by_id(uid=uid)
 
     async def create(self, data: dict[str, Any] | UserNotification) -> UserNotification:
         return await self.user_notification_repository.create(data=data)
 
-    async def update(self, id_value: int, data: dict[str, Any]) -> UserNotification:
-        return await self.user_notification_repository.update(id_value=id_value, data=data)
+    async def update(self, uid: int, data: dict[str, Any]) -> UserNotification:
+        return await self.user_notification_repository.update(uid=uid, data=data)
 
     async def one(
         self,
@@ -41,10 +41,10 @@ async def all(
             filters=filters, order_by=order_by, pagination=pagination, pager=pager
         )
 
-    async def new_by_user_id(self, id_value: int | list[int]) -> Sequence[UserNotification]:
+    async def new_by_user_id(self, uid: int | list[int]) -> Sequence[UserNotification]:
         return await self.all(  # type: ignore
             filters=[
-                Filter("user_id", FilterOperator.IN if isinstance(id_value, list) else FilterOperator.EQ, id_value),
-                Filter("status", FilterOperator.EQ, UserNotificationStatus.NEW),
+                Filter("user_id", Oper.IN if isinstance(uid, list) else Oper.EQ, uid),
+                Filter("status", Oper.EQ, UserNotificationStatus.NEW),
             ],
         )

src/app/user_notification/service/ws_notification_service.py

@@ -0,0 +1,83 @@
+from datetime import datetime
+from typing import Any
+
+from fastapi import WebSocket
+
+from src.app.user_notification.data.user_notification_status import UserNotificationStatus
+from src.app.user_notification.service.user_notification_service import UserNotificationService
+from src.app.ws.service.ws_handler import WSHandler
+from src.core.log.log import Log
+from src.core.web_socket.enum.ws_type import WSType
+from src.core.web_socket.ws_manager import WSManager
+
+
+class WSNotificationService(WSHandler):
+    def __init__(
+            self,
+            user_notification_service: UserNotificationService,
+            ws_manager: WSManager,
+            log: Log,
+    ) -> None:
+        self.user_notification_service = user_notification_service
+        self.ws_manager = ws_manager
+        self.log = log
+
+    async def add_connection(self, user_id: str, websocket: WebSocket) -> None:
+        notifications = await self.user_notification_service.new_by_user_id(uid=int(user_id))
+        for notification in notifications:
+            await self.ws_manager.send_to_user(
+                user_id=user_id,
+                data={
+                    "type": WSType.USER_NOTIFICATION.value,
+                    "data": {
+                        "message": notification.data.get("message"),
+                        "status": notification.status.value,
+                        "createdAt": notification.created_at.isoformat() if isinstance(notification.created_at, datetime) else str(notification.created_at),
+                    }
+                },
+                websocket=websocket,
+            )
+        self.log.info(f"WS connect: user {user_id}, {len(notifications)} notifications sent")
+
+    async def process_message(self, user_id: str,message: dict[str, Any], websocket: WebSocket) -> None:
+        message_type = WSType(message.get("type", WSType.UNKNOWN.value))
+        if message_type == WSType.USER_NOTIFICATION:
+            await self.user_notification_service.create(
+                data={
+                    "user_id": int(user_id),
+                    "message": message.get("message"),
+                    "status": UserNotificationStatus.NEW,
+                },
+            )
+            return
+        if message_type == WSType.MESSAGE_READ:
+            uid = message.get("id")
+            if uid is None:
+                self.log.warning(f"WS message: user {user_id}, message_read message without id: {message}")
+                return
+            notification = await self.user_notification_service.update(
+                uid=int(uid),
+                data={
+                    "status": UserNotificationStatus.READ,
+                },
+            )
+            await self.ws_manager.send_to_user(
+                user_id=user_id,
+                data={
+                    "type": WSType.USER_NOTIFICATION.value,
+                    "data": {
+                        "message": notification.data.get("message"),
+                        "status": notification.status.value,
+                        "createdAt": notification.created_at.isoformat() if isinstance(notification.created_at, datetime) else str(notification.created_at),
+                    }
+                },
+            )
+            return
+
+
+    async def remove_connection(self, user_id: str, websocket: WebSocket) -> None:
+        pass
+
+    @staticmethod
+    def can(ws_type: WSType) -> bool:
+        return ws_type == WSType.USER_NOTIFICATION