Add user register API endpoint

migrations/versions/8652e8501339_add_user_and_slack_team.py

@@ -0,0 +1,50 @@
+"""add user and slack team
+
+Revision ID: 8652e8501339
+Revises: 
+Create Date: 2024-10-24 16:52:59.936776
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '8652e8501339'
+down_revision: Union[str, None] = None
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('slack_teams',
+    sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+    sa.Column('name', sa.String(), nullable=False),
+    sa.Column('slack_team_id', sa.String(), nullable=False),
+    sa.Column('token', sa.String(), nullable=False),
+    sa.Column('created_at', sa.DateTime(), nullable=False),
+    sa.PrimaryKeyConstraint('id')
+    )
+    op.create_table('users',
+    sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+    sa.Column('name', sa.String(), nullable=False),
+    sa.Column('content_delivery_frequency', sa.Enum('DAILY', 'WEEKLY', 'BI_WEEKLY', 'MONTHLY', name='contentdeliveryfrequency'), nullable=False),
+    sa.Column('slack_id', sa.String(), nullable=True),
+    sa.Column('slack_team_id', sa.Integer(), nullable=True),
+    sa.Column('email_id', sa.String(), nullable=False),
+    sa.Column('last_message_sent_at', sa.DateTime(), nullable=True),
+    sa.Column('created_at', sa.DateTime(), nullable=False),
+    sa.ForeignKeyConstraint(['slack_team_id'], ['slack_teams.id'], ),
+    sa.PrimaryKeyConstraint('id')
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table('users')
+    op.drop_table('slack_teams')
+    # ### end Alembic commands ###

src/app.py

@@ -9,7 +9,11 @@ from fastapi.middleware.cors import CORSMiddleware
 
 from src.services import ContentDeliveryService
 from src.utils import logger
-from src.controllers import bot_router
+from src.controllers import bot_router, user_router
+
+from alembic import command
+from alembic.config import Config
+from src.repositories import DatabaseConfig
 
 scheduler = AsyncIOScheduler()
 
@@ -23,6 +27,21 @@ async def scheduled_job():
         logger.error(e)
 
 
+def run_upgrade(connection, alembic_config: Config):
+    alembic_config.attributes["connection"] = connection
+    command.upgrade(alembic_config, "head")
+
+
+async def run_migrations():
+    logger.info("Running migrations if any...")
+    alembic_config = Config("alembic.ini")
+    alembic_config.set_main_option(
+        "sqlalchemy.url", os.getenv("SQLALCHEMY_DATABASE_URI")
+    )
+    async with DatabaseConfig.async_engine().begin() as session:
+        await session.run_sync(run_upgrade, alembic_config)
+
+
 def run_scheduler():
     loop = asyncio.new_event_loop()
     asyncio.set_event_loop(loop)
@@ -34,6 +53,7 @@ def run_scheduler():
 async def lifespan(app: FastAPI):
     try:
         logger.info("Starting up the application...")
+        await run_migrations()
         scheduler_thread = Thread(target=run_scheduler, daemon=True)
         scheduler_thread.start()
         logger.info("Application started successfully...")
@@ -63,3 +83,4 @@ async def check_health():
     return {"response": "Service is healthy!"}
 
 app.include_router(bot_router, prefix="/api/v1")
+app.include_router(user_router, prefix="/api/v1")

src/controllers/__init__.py

@@ -1,7 +1,9 @@
 from ._bot_controller import BotController
+from ._user_controller import UserController
 
 bot_router = BotController().router
+user_router = UserController().router
 
-__all__ = ["bot_router"]
+__all__ = ["bot_router", "user_router"]
 __version__ = "0.1.0"
 __author__ = "Ashok Bhati"

src/controllers/_user_controller.py

@@ -0,0 +1,28 @@
+import os
+from fastapi import APIRouter, HTTPException
+from src.utils import logger
+from src.services import UserService
+
+
+class UserController:
+    def __init__(self):
+        self.service = UserService
+        self.router = APIRouter(prefix="/users", tags=["User"])
+        self.router.add_api_route("/", self.registerUser, methods=["POST"])
+
+    async def registerUser(self, user: dict):
+        logger.info(f"Received event: {user}")
+        try:
+            if not "email_id" in user:
+                raise HTTPException(status_code=400, detail="Missing email id")
+            if not "name" in user:
+                raise HTTPException(status_code=400, detail="Missing name")
+            async with self.service() as service:
+                user = await service.register_user(user)
+                return {"message": "User registered successfully", "user": user}
+        except HTTPException as e:
+            logger.warning(e)
+            raise e
+        except Exception as e:
+            logger.error(e)
+            raise HTTPException(status_code=500, detail=str(e))

src/models/__init__.py

@@ -0,0 +1,7 @@
+from ._base import Base
+from ._user import User, ContentDeliveryFrequency
+from ._slack_team import SlackTeam
+
+__all__ = ["Base", "User", "ContentDeliveryFrequency", "SlackTeam"]
+__version__ = "0.1.0"
+__author__ = "Kanha Upadhyay"

src/models/_base.py

@@ -0,0 +1,3 @@
+from sqlalchemy.orm import declarative_base
+
+Base = declarative_base()

src/models/_slack_team.py

@@ -0,0 +1,28 @@
+from enum import Enum as PyEnum
+
+from sqlalchemy import Column, DateTime, Enum, Integer, String, func
+from sqlalchemy.orm import relationship
+
+from ._base import Base
+
+
+class ContentDeliveryFrequency(PyEnum):
+    DAILY = "DAILY"
+    WEEKLY = "WEEKLY"
+    BI_WEEKLY = "BI_WEEKLY"
+    MONTHLY = "MONTHLY"
+
+
+class SlackTeam(Base):
+    __tablename__ = "slack_teams"
+
+    id = Column(Integer, primary_key=True, autoincrement=True, nullable=False)
+    name = Column(String, nullable=False)
+    slack_team_id = Column(
+        String,
+        nullable=False,
+    )
+    token = Column(String, nullable=False)
+    created_at = Column(DateTime, nullable=False, default=func.now())
+
+    users = relationship("User", back_populates="slack_team")

src/models/_user.py

@@ -0,0 +1,36 @@
+from enum import Enum as PyEnum
+
+from sqlalchemy import Column, DateTime, Enum, Integer, String, ForeignKey, func
+from sqlalchemy.orm import relationship
+
+from ._base import Base
+
+
+class ContentDeliveryFrequency(PyEnum):
+    DAILY = "DAILY"
+    WEEKLY = "WEEKLY"
+    BI_WEEKLY = "BI_WEEKLY"
+    MONTHLY = "MONTHLY"
+
+
+class User(Base):
+    __tablename__ = "users"
+
+    id = Column(Integer, primary_key=True, autoincrement=True, nullable=False)
+    name = Column(String, nullable=False)
+    content_delivery_frequency = Column(
+        Enum(ContentDeliveryFrequency),
+        nullable=False,
+        default=ContentDeliveryFrequency.WEEKLY,
+    )
+    slack_id = Column(String, nullable=True)
+    slack_team_id = Column(
+        Integer,
+        ForeignKey("slack_teams.id"),
+        nullable=True,
+    )
+    email_id = Column(String, nullable=False)
+    last_message_sent_at = Column(DateTime, nullable=True)
+    created_at = Column(DateTime, nullable=False, default=func.now())
+
+    slack_team = relationship("SlackTeam", back_populates="users")

src/repositories/__init__.py

@@ -0,0 +1,8 @@
+from ._base_repository import BaseRepository
+from ._config import DatabaseConfig
+from ._user_repository import UserRepository
+from ._slack_team_repository import SlackTeamRepository
+
+__all__ = ["DatabaseConfig", "BaseRepository", "UserRepository", "SlackTeamRepository"]
+__version__ = "0.1.0"
+__author__ = "Kanha Upadhyay"

src/repositories/_base_repository.py

@@ -0,0 +1,81 @@
+from contextlib import asynccontextmanager
+
+from sqlalchemy.future import select
+
+from ._config import DatabaseConfig
+
+
+class BaseRepository:
+    def __init__(self, model=None):
+        self.__model = model
+        assert self.__model is not None, "model must be defined"
+
+    @asynccontextmanager
+    async def get_session(self):
+        async with DatabaseConfig.async_session() as session:
+            yield session
+
+    async def execute(self, query):
+        async with self.get_session() as session:
+            result = await session.execute(query)
+            return result
+
+    async def create(self, kwargs: dict):
+        async with self.get_session() as session:
+            instance = self.__model(**kwargs)
+            session.add(instance)
+            await session.commit()
+            await session.refresh(instance)
+            return instance
+
+    async def get_all(
+        self,
+        page: int = 1,
+        page_size: int = 10,
+        filter_by: dict = None,
+        order_by: dict = None,
+    ):
+        async with self.get_session() as session:
+            offset = (page - 1) * page_size
+            query = select(self.__model).offset(offset).limit(page_size)
+            if filter_by:
+                query = query.filter_by(**filter_by)
+            if order_by:
+                for field, direction in order_by.items():
+                    if direction == "asc":
+                        query = query.order_by(getattr(self.__model, field).asc())
+                    elif direction == "desc":
+                        query = query.order_by(getattr(self.__model, field).desc())
+            result = await session.execute(query)
+            return [row for row in result.scalars()]
+
+    async def get(self, id, filter_by: dict = None):
+        async with self.get_session() as session:
+            query = select(self.__model).where(self.__model.id == id)
+            if filter_by:
+                query = query.filter_by(**filter_by)
+            result = await session.execute(query)
+            result = result.scalar()
+            if result:
+                return result
+            raise Exception(f"{self.__model.__name__} not found")
+
+    async def patch(self, id, **kwargs: dict):
+        async with self.get_session() as session:
+            instance = await session.get(self.__model, id)
+            if instance:
+                for key, value in kwargs.items():
+                    setattr(instance, key, value)
+                await session.commit()
+                await session.refresh(instance)
+                return instance
+            raise Exception(f"{self.__model.__name__} not found")
+
+    async def delete(self, id):
+        async with self.get_session() as session:
+            instance = await session.get(self.__model, id)
+            if instance:
+                await session.delete(instance)
+                await session.commit()
+                return True
+            raise Exception(f"{self.__model.__name__} not found")

src/repositories/_config.py

@@ -0,0 +1,29 @@
+import os
+from contextlib import asynccontextmanager
+
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+
+class DatabaseConfig:
+    @classmethod
+    def async_engine(cls):
+        return create_async_engine(
+            os.getenv("SQLALCHEMY_DATABASE_URI"), pool_pre_ping=True, pool_recycle=3600
+        )
+
+    @classmethod
+    @asynccontextmanager
+    async def async_session(cls):
+        engine = cls.async_engine()
+        session_factory = async_sessionmaker(
+            bind=engine,
+            autoflush=False,
+            autocommit=False,
+            expire_on_commit=False,
+        )
+        session = session_factory()
+        try:
+            yield session
+        finally:
+            await session.close()
+            await engine.dispose()

src/repositories/_slack_team_repository.py

@@ -0,0 +1,8 @@
+from src.models import SlackTeam
+
+from ._base_repository import BaseRepository
+
+
+class SlackTeamRepository(BaseRepository):
+    def __init__(self):
+        super().__init__(model=SlackTeam)

src/repositories/_user_repository.py

@@ -0,0 +1,16 @@
+from sqlalchemy import select
+
+from src.models import User
+
+from ._base_repository import BaseRepository
+
+
+class UserRepository(BaseRepository):
+    def __init__(self):
+        super().__init__(model=User)
+
+    async def get_user_by_email(self, email: str):
+        query = select(User).where(User.email == email)
+        result = await self.execute(query)
+        result = result.scalar_one_or_none()
+        return result

src/services/__init__.py

@@ -1,5 +1,6 @@
 from ._content_delivery_service import ContentDeliveryService
+from ._user_service import UserService
 
-__all__ = ["ContentDeliveryService"]
+__all__ = ["ContentDeliveryService", "UserService"]
 __version__ = "0.1.0"
 __author__ = "Ashok Bhati"

src/services/_slack_team_service.py

@@ -0,0 +1,23 @@
+from src.repositories import SlackTeamRepository
+
+from src.models import ContentDeliveryFrequency
+import datetime
+
+
+class SlackTeamService:
+    def __init__(self):
+        self.slackTeam_repository = SlackTeamRepository()
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        pass
+
+    async def register_slackTeam(self, email: str):
+        return await self.slackTeam_repository.create(email=email)
+
+    async def get_slackTeam_token(self, slackTeam_id):
+        return await self.slackTeam_repository.get(
+            slackTeam_id, filter_by={"token": True}
+        )

src/services/_user_service.py

@@ -0,0 +1,48 @@
+from src.repositories import UserRepository
+
+from src.models import ContentDeliveryFrequency
+import datetime
+
+
+class UserService:
+    def __init__(self):
+        self.user_repository = UserRepository()
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        pass
+
+    async def register_user(self, user: dict) -> dict:
+        return await self.user_repository.create(user)
+
+    async def get_user_info_by_email(self, email: str):
+        return await self.user_repository.get_user_by_email(email=email)
+
+    async def update_user_slack_info(self, user_id, info):
+        return await self.user_repository.patch(user_id, info)
+
+    async def update_user_frequency(self, user_id, frequency: ContentDeliveryFrequency):
+        return await self.user_repository.patch(
+            user_id, {"content_delivery_frequency": frequency.value}
+        )
+
+    async def get_users_list(self):
+        return await self.user_repository.get_all()
+
+    async def get_user_last_message_time(self, user_id):
+        return await self.user_repository.get(
+            user_id, filter_by={"last_message_time": True}
+        )
+
+    async def update_user_last_message_time(self, user_id):
+        return await self.user_repository.patch(
+            id=user_id,
+            last_message_time=datetime.now(),
+        )
+
+    async def get_user_frequency(self, user_id):
+        return await self.user_repository.get(
+            user_id, filter_by={"content_delivery_frequency": True}
+        )