posts crud and routes

src/app/api/v1/__init__.py

@@ -2,7 +2,9 @@
 
 from app.api.v1.login import router as login_router
 from app.api.v1.users import router as users_router
+from app.api.v1.posts import router as posts_router
 
 router = APIRouter(prefix="/v1")
 router.include_router(login_router)
 router.include_router(users_router)
+router.include_router(posts_router)

src/app/api/v1/posts.py

@@ -0,0 +1,139 @@
+from typing import List, Annotated
+
+from fastapi import Depends, HTTPException
+from sqlalchemy.ext.asyncio import AsyncSession
+import fastapi
+
+from app.schemas.post import PostCreate, PostUpdate, PostRead, PostCreateInternal
+from app.schemas.user import UserRead
+from app.api.dependencies import get_current_user, get_current_superuser
+from app.core.database import async_get_db
+from app.crud.crud_posts import crud_posts
+from app.crud.crud_users import crud_users
+from app.api.exceptions import privileges_exception
+
+router = fastapi.APIRouter(tags=["posts"])
+
+@router.post("/{username}/post", response_model=PostRead, status_code=201)
+async def write_post(
+    username: str,
+    post: PostCreate, 
+    current_user: Annotated[UserRead, Depends(get_current_user)],
+    db: Annotated[AsyncSession, Depends(async_get_db)]
+):
+    db_user = await crud_users.get(db=db, username=username, is_deleted=False)
+    if db_user is None:
+        raise HTTPException(status_code=404, detail="User not found")
+
+    if current_user.id != db_user.id:
+        raise privileges_exception
+
+    post_internal_dict = post.model_dump()
+    post_internal_dict["created_by_user_id"] = db_user.id
+    post_internal = PostCreateInternal(**post_internal_dict)
+    return await crud_posts.create(db=db, object=post_internal)
+
+
+@router.get("/{username}/posts", response_model=List[PostRead])
+async def read_posts(
+    username: str, 
+    db: Annotated[AsyncSession, Depends(async_get_db)]
+):
+    db_user = await crud_users.get(db=db, username=username, is_deleted=False)
+    if db_user is None:
+        raise HTTPException(status_code=404, detail="User not found")
+
+    posts = await crud_posts.get_multi(db=db, created_by_user_id=db_user.id, is_deleted=False)
+    return posts
+
+
+@router.get("/{username}/post/{id}", response_model=PostRead)
+async def read_post(
+    username: str,
+    id: int, 
+    db: Annotated[AsyncSession, Depends(async_get_db)]
+):
+    db_user = await crud_users.get(db=db, username=username, is_deleted=False)
+    if db_user is None:
+        raise HTTPException(status_code=404, detail="User not found")
+
+    db_post = await crud_posts.get(db=db, id=id, created_by_user_id=db_user.id, is_deleted=False)
+    if db_post is None:
+        raise HTTPException(status_code=404, detail="Post not found")
+
+    return db_post
+
+
+@router.patch("/{username}/post/{id}", response_model=PostRead)
+async def patch_post(
+    username: str,
+    id: int,
+    values: PostUpdate,
+    current_user: Annotated[UserRead, Depends(get_current_user)],
+    db: Annotated[AsyncSession, Depends(async_get_db)]
+):
+    db_user = await crud_users.get(db=db, username=username, is_deleted=False)
+    if db_user is None:
+        raise HTTPException(status_code=404, detail="User not found")
+
+    if current_user.id != db_user.id:
+        raise privileges_exception
+
+    db_post = await crud_posts.get(db=db, id=id, is_deleted=False)
+    if db_post is None:
+        raise HTTPException(status_code=404, detail="Post not found")
+
+    return await crud_posts.update(db=db, object=values, db_object=db_post, id=id)
+
+
+@router.delete("/{username}/post/{id}")
+async def erase_post(
+    username: str,
+    id: int,
+    current_user: Annotated[UserRead, Depends(get_current_user)],
+    db: Annotated[AsyncSession, Depends(async_get_db)]
+):
+    db_user = await crud_users.get(db=db, username=username, is_deleted=False)
+    if db_user is None:
+        raise HTTPException(status_code=404, detail="User not found")
+
+    if current_user.id != db_user.id:
+        raise privileges_exception
+
+    db_post = await crud_posts.get(db=db, id=id, is_deleted=False)
+    if db_post is None:
+        raise HTTPException(status_code=404, detail="Post not found")
+
+    deleted_post = await crud_posts.delete(db=db, db_object=db_post, id=id)
+    if deleted_post.is_deleted == True:
+        message = {"message": "Post deleted"}
+    else:
+        message = {"message": "Something went wrong"}
+
+    return message
+
+
+@router.delete("/{username}/db_post/{id}")
+async def erase_db_post(
+    username: str,
+    id: int,
+    current_superuser: Annotated[UserRead, Depends(get_current_superuser)],
+    db: Annotated[AsyncSession, Depends(async_get_db)]
+):
+    db_user = await crud_users.get(db=db, username=username, is_deleted=False)
+    if db_user is None:
+        raise HTTPException(status_code=404, detail="User not found")
+
+    db_post = await crud_posts.get(db=db, id=id, is_deleted=False)
+    if db_post is None:
+        raise HTTPException(status_code=404, detail="Post not found")
+
+    await crud_posts.db_delete(db=db, db_object=db_post, id=id)
+    deleted_post = await crud_posts.get(db=db, id=id)
+
+    if deleted_post is None:
+        message = {"message": "Post deleted"}
+    else:
+        message = {"message": "Something went wrong"}
+
+    return message

src/app/api/v1/users.py

@@ -13,7 +13,7 @@
 
 router = fastapi.APIRouter(tags=["users"])
 
-@router.post("/user", response_model=UserBase, status_code=201)
+@router.post("/user", response_model=UserRead, status_code=201)
 async def write_user(user: UserCreate, db: AsyncSession = Depends(async_get_db)):
     db_user = await crud_users.get(db=db, email=user.email)
     if db_user:
@@ -53,7 +53,7 @@ async def read_user(username: str, db: AsyncSession = Depends(async_get_db)):
     return db_user
 
 
-@router.patch("/user/{username}", response_model=UserUpdate)
+@router.patch("/user/{username}", response_model=UserRead)
 async def patch_user(
     values: UserUpdate,
     username: str,

src/app/crud/crud_posts.py

@@ -0,0 +1,6 @@
+from app.crud.crud_base import CRUDBase
+from app.models.post import Post
+from app.schemas.post import PostCreateInternal, PostUpdate, PostUpdateInternal, PostDelete
+
+CRUDPost = CRUDBase[Post, PostCreateInternal, PostUpdate, PostUpdateInternal, PostDelete]
+crud_posts = CRUDPost(Post)

src/app/models/post.py

@@ -16,9 +16,9 @@ class Post(Base):
     created_by_user_id: Mapped[int] = mapped_column(ForeignKey("user.id"))
     title: Mapped[str] = mapped_column(String(30))
     text: Mapped[str] = mapped_column(String(63206))
-    media_url: Mapped[str] = mapped_column(String)
+    media_url: Mapped[str | None] = mapped_column(String, default=None)
 
-    user: Mapped["User"] = relationship(back_populates="posts", lazy="selectin")
+    user: Mapped["User"] = relationship(back_populates="posts", lazy="selectin", init=False)
 
     created_at: Mapped[datetime] = mapped_column(
         DateTime, default_factory=datetime.utcnow

src/app/schemas/post.py

@@ -41,7 +41,6 @@ class PostRead(BaseModel):
     media_url: Annotated[
         str | None, 
         Field(
-            pattern=r"^(https?|ftp)://[^\s/$.?#].[^\s]*$",
             examples=["https://www.postimageurl.com"],
             default=None
         ),
@@ -53,8 +52,18 @@ class PostRead(BaseModel):
 class PostCreate(PostBase):
     model_config = ConfigDict(extra='forbid')
 
+    media_url: Annotated[
+        str | None, 
+        Field(
+            pattern=r"^(https?|ftp)://[^\s/$.?#].[^\s]*$",
+            examples=["https://www.postimageurl.com"],
+            default=None
+        ),
+    ]
+
+
+class PostCreateInternal(PostCreate):
     created_by_user_id: int
-    media_url: str | None = None
 
 
 class PostUpdate(PostBase):

src/migrations/versions/474ddfe32ea7_.py

@@ -0,0 +1,36 @@
+"""empty message
+
+Revision ID: 474ddfe32ea7
+Revises: 211a8b0d0080
+Create Date: 2023-10-07 02:22:22.612109
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '474ddfe32ea7'
+down_revision: Union[str, None] = '211a8b0d0080'
+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.alter_column('post', 'media_url',
+               existing_type=sa.VARCHAR(),
+               nullable=True)
+    op.create_unique_constraint(None, 'post', ['id'])
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint(None, 'post', type_='unique')
+    op.alter_column('post', 'media_url',
+               existing_type=sa.VARCHAR(),
+               nullable=False)
+    # ### end Alembic commands ###