python mongodb crud app

Author: wpcodevo

app/database.py

@@ -7,4 +7,6 @@
 
 db = client[settings.MONGO_INITDB_DATABASE]
 User = db.users
+Post = db.posts
 User.create_index([("email", pymongo.ASCENDING)], unique=True)
+Post.create_index([("title", pymongo.ASCENDING)], unique=True)

app/main.py

@@ -2,7 +2,7 @@
 from fastapi.middleware.cors import CORSMiddleware
 
 from app.config import settings
-from app.routers import auth, user
+from app.routers import auth, user, post
 
 app = FastAPI()
 
@@ -21,6 +21,7 @@
 
 app.include_router(auth.router, tags=['Auth'], prefix='/api/auth')
 app.include_router(user.router, tags=['Users'], prefix='/api/users')
+app.include_router(post.router, tags=['Posts'], prefix='/api/posts')
 
 
 @app.get("/api/healthchecker")

app/oauth2.py

@@ -5,7 +5,7 @@
 from pydantic import BaseModel
 from bson.objectid import ObjectId
 
-from app.serializers import userEntity
+from app.serializers.userSerializers import userEntity
 
 from .database import User
 from .config import settings

app/routers/auth.py

@@ -8,7 +8,7 @@
 from app import oauth2
 from app.database import User
 from app.email import Email
-from app.serializers import userEntity
+from app.serializers.userSerializers import userEntity
 from .. import schemas, utils
 from app.oauth2 import AuthJWT
 from ..config import settings

app/routers/post.py

@@ -0,0 +1,87 @@
+from datetime import datetime
+from fastapi import Depends, HTTPException, status, APIRouter, Response
+from pymongo.collection import ReturnDocument
+from app import schemas
+from app.database import Post
+from app.oauth2 import require_user
+from app.serializers.postSerializers import postEntity, postListEntity
+from bson.objectid import ObjectId
+
+router = APIRouter()
+
+
+@router.get('/')
+def get_posts(limit: int = 10, page: int = 1, search: str = '', user_id: str = Depends(require_user)):
+    skip = (page - 1) * limit
+    pipeline = [
+        {'$match': {}},
+        {'$lookup': {'from': 'users', 'localField': 'user',
+                     'foreignField': '_id', 'as': 'user'}},
+        {'$unwind': '$user'},
+        {
+            '$skip': skip
+        }, {
+            '$limit': limit
+        }
+    ]
+    posts = postListEntity(Post.aggregate(pipeline))
+    return {'status': 'success', 'results': len(posts), 'posts': posts}
+
+
+@router.post('/', status_code=status.HTTP_201_CREATED)
+def create_post(post: schemas.CreatePostSchema, user_id: str = Depends(require_user)):
+    post.user = ObjectId(user_id)
+    post.created_at = datetime.utcnow()
+    post.updated_at = post.created_at
+    result = Post.insert_one(post.dict())
+    pipeline = [
+        {'$match': {'_id': result.inserted_id}},
+        {'$lookup': {'from': 'users', 'localField': 'user',
+                     'foreignField': '_id', 'as': 'user'}},
+        {'$unwind': '$user'},
+    ]
+    new_post = postListEntity(Post.aggregate(pipeline))[0]
+    return new_post
+
+
+@router.put('/{id}')
+def update_post(id: str, payload: schemas.UpdatePostSchema, user_id: str = Depends(require_user)):
+    if not ObjectId.is_valid(id):
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
+                            detail=f"Invalid id: {id}")
+    updated_post = Post.find_one_and_update(
+        {'_id': ObjectId(id)}, {'$set': payload.dict(exclude_none=True)}, return_document=ReturnDocument.AFTER)
+    if not updated_post:
+        raise HTTPException(status_code=status.HTTP_200_OK,
+                            detail=f'No post with this id: {id} found')
+    return postEntity(updated_post)
+
+
+@router.get('/{id}')
+def get_post(id: str, user_id: str = Depends(require_user)):
+    if not ObjectId.is_valid(id):
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
+                            detail=f"Invalid id: {id}")
+    pipeline = [
+        {'$match': {'_id': ObjectId(id)}},
+        {'$lookup': {'from': 'users', 'localField': 'user',
+                     'foreignField': '_id', 'as': 'user'}},
+        {'$unwind': '$user'},
+    ]
+    post = postListEntity(Post.aggregate(pipeline))[0]
+    if not post:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
+                            detail=f"No post with this id: {id} found")
+    return post
+
+
+@router.delete('/{id}')
+def delete_post(id: str, user_id: str = Depends(require_user)):
+    if not ObjectId.is_valid(id):
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
+                            detail=f"Invalid id: {id}")
+    post = Post.find_one_and_delete({'_id': ObjectId(id)})
+    if not post:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
+                            detail=f'No post with this id: {id} found')
+    return Response(status_code=status.HTTP_204_NO_CONTENT)

app/routers/user.py

@@ -1,6 +1,6 @@
 from fastapi import APIRouter, Depends
 from bson.objectid import ObjectId
-from app.serializers import userResponseEntity
+from app.serializers.userSerializers import userResponseEntity
 
 from app.database import User
 from .. import schemas, oauth2

app/schemas.py

@@ -1,5 +1,7 @@
 from datetime import datetime
+from typing import List
 from pydantic import BaseModel, EmailStr, constr
+from bson.objectid import ObjectId
 
 
 class UserBaseSchema(BaseModel):
@@ -33,3 +35,54 @@ class UserResponseSchema(UserBaseSchema):
 class UserResponse(BaseModel):
     status: str
     user: UserResponseSchema
+
+
+class FilteredUserResponse(UserBaseSchema):
+    id: str
+
+
+class PostBaseSchema(BaseModel):
+    title: str
+    content: str
+    category: str
+    image: str
+    created_at: datetime | None = None
+    updated_at: datetime | None = None
+
+    class Config:
+        orm_mode = True
+        allow_population_by_field_name = True
+        arbitrary_types_allowed = True
+        json_encoders = {ObjectId: str}
+
+
+class CreatePostSchema(PostBaseSchema):
+    user: ObjectId | None = None
+    pass
+
+
+class PostResponse(PostBaseSchema):
+    id: str
+    user: FilteredUserResponse
+    created_at: datetime
+    updated_at: datetime
+
+
+class UpdatePostSchema(BaseModel):
+    title: str | None = None
+    content: str | None = None
+    category: str | None = None
+    image: str | None = None
+    user: str | None = None
+
+    class Config:
+        orm_mode = True
+        allow_population_by_field_name = True
+        arbitrary_types_allowed = True
+        json_encoders = {ObjectId: str}
+
+
+class ListPostResponse(BaseModel):
+    status: str
+    results: int
+    posts: List[PostResponse]

app/serializers/postSerializers.py

@@ -0,0 +1,31 @@
+from app.serializers.userSerializers import embeddedUserResponse
+
+
+def postEntity(post) -> dict:
+    return {
+        "id": str(post["_id"]),
+        "title": post["title"],
+        "category": post["category"],
+        "content": post["content"],
+        "image": post["image"],
+        "user": str(post["user"]),
+        "created_at": post["created_at"],
+        "updated_at": post["updated_at"]
+    }
+
+
+def populatedPostEntity(post) -> dict:
+    return {
+        "id": str(post["_id"]),
+        "title": post["title"],
+        "category": post["category"],
+        "content": post["content"],
+        "image": post["image"],
+        "user": embeddedUserResponse(post["user"]),
+        "created_at": post["created_at"],
+        "updated_at": post["updated_at"]
+    }
+
+
+def postListEntity(posts) -> list:
+    return [populatedPostEntity(post) for post in posts]

app/serializers.py renamed to app/serializers/userSerializers.py

@@ -24,5 +24,14 @@ def userResponseEntity(user) -> dict:
     }
 
 
+def embeddedUserResponse(user) -> dict:
+    return {
+        "id": str(user["_id"]),
+        "name": user["name"],
+        "email": user["email"],
+        "photo": user["photo"]
+    }
+
+
 def userListEntity(users) -> list:
     return [userEntity(user) for user in users]