feat: created admin

identity_socializer/db/dao/admin_dao.py

@@ -0,0 +1,42 @@
+from typing import List
+
+from fastapi import Depends
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from identity_socializer.db.dependencies import get_db_session
+from identity_socializer.db.models.admin_model import AdminModel
+
+
+class AdminDAO:
+    """Class for accessing admins table."""
+
+    def __init__(self, session: AsyncSession = Depends(get_db_session)):
+        self.session = session
+
+    async def create_admin_model(
+        self,
+        admin_id: str,
+        email: str,
+    ) -> None:
+        """Add single admin to session."""
+        admin_model = AdminModel(
+            id=admin_id,
+            email=email,
+        )
+
+        self.session.add(admin_model)
+
+    async def get_all_admins(self, limit: int, offset: int) -> List[AdminModel]:
+        """
+        Get all admins models with limit/offset pagination.
+
+        :param limit: limit of admins.
+        :param offset: offset of admins.
+        :return: stream of admins.
+        """
+        raw_admins = await self.session.execute(
+            select(AdminModel).limit(limit).offset(offset),
+        )
+
+        return list(raw_admins.scalars().fetchall())

identity_socializer/db/migrations/versions/2023-10-19-14-45_22ac5d68869b.py

@@ -0,0 +1,31 @@
+"""create admin table
+
+Revision ID: 22ac5d68869b
+Revises: 847eee4ea866
+Create Date: 2023-10-19 14:45:52.775871
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "22ac5d68869b"
+down_revision = "847eee4ea866"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    length = 200
+
+    op.create_table(
+        "admins",
+        sa.Column("id", sa.String(length), nullable=False),
+        sa.Column("email", sa.String(length), nullable=False),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("admins")

identity_socializer/db/models/admin_model.py

@@ -0,0 +1,21 @@
+import datetime
+
+from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy.sql.sqltypes import DateTime, String
+
+from identity_socializer.db.base import Base
+
+
+class AdminModel(Base):
+    """Model for Admin."""
+
+    __tablename__ = "admins"
+
+    length = 200
+
+    id: Mapped[str] = mapped_column(primary_key=True)
+    email: Mapped[str] = mapped_column(String(length))
+    created_at: Mapped[DateTime] = mapped_column(
+        DateTime,
+        default=datetime.datetime.utcnow,
+    )

identity_socializer/web/api/auth/schema.py

@@ -34,3 +34,11 @@ class SimpleUserModelDTO(BaseModel):
     name: str
     email: str
     model_config = ConfigDict(from_attributes=True)
+
+
+class AdminDTO(BaseModel):
+    """Message model for admin register."""
+
+    id: str
+    email: str
+    model_config = ConfigDict(from_attributes=True)

identity_socializer/web/api/auth/views.py

@@ -3,9 +3,12 @@
 from fastapi import APIRouter, Depends, HTTPException
 from firebase_admin import auth
 
+from identity_socializer.db.dao.admin_dao import AdminDAO
 from identity_socializer.db.dao.user_dao import UserDAO
+from identity_socializer.db.models.admin_model import AdminModel
 from identity_socializer.db.models.user_model import UserModel
 from identity_socializer.web.api.auth.schema import (
+    AdminDTO,
     SecurityToken,
     SimpleUserModelDTO,
     Success,
@@ -64,3 +67,25 @@ async def get_user_models(
     :return: list of users objects from database.
     """
     return await user_dao.get_all_users(limit=limit, offset=offset)
+
+
+@router.post("/create_admin", response_model=None)
+async def create_admin(
+    admin: AdminDTO,
+    admin_dao: AdminDAO = Depends(),
+) -> None:
+    """Create an admin."""
+    await admin_dao.create_admin_model(
+        admin_id=admin.id,
+        email=admin.email,
+    )
+
+
+@router.get("/admins", response_model=None)
+async def get_admin_models(
+    limit: int = 10,
+    offset: int = 0,
+    user_dao: AdminDAO = Depends(),
+) -> List[AdminModel]:
+    """Retrieve all admins from the database."""
+    return await user_dao.get_all_admins(limit=limit, offset=offset)