Update generated backend for blog_app with entities: posts, comments, tags, user

app/api/core/dependencies/__init__.py

@@ -0,0 +1 @@
+from .dependencies import get_db

app/api/core/dependencies/dependencies.py

@@ -1,3 +1,4 @@
+# app/api/core/dependencies/dependencies.py
 from sqlalchemy.orm import Session
 from app.api.db.database import SessionLocal
 

app/api/core/middleware/activity_tracker.py

@@ -9,5 +9,5 @@ class ActivityTrackerMiddleware(BaseHTTPMiddleware):
         start_time = time.time()
         response: Response = await call_next(request)
         process_time = time.time() - start_time
-        logger.info(f"Request processed in {process_time:.4f} seconds")
+        logger.info(f"Processed {request.method} {request.url} in {process_time:.4f} seconds")
         return response

app/api/utils/__init__.py

@@ -0,0 +1 @@
+from .utils import get_current_timestamp

app/api/utils/utils.py

@@ -0,0 +1,4 @@
+import datetime
+
+def get_current_timestamp():
+    return datetime.datetime.utcnow().isoformat()

app/api/v1/models/comments.py

@@ -1,21 +1,34 @@
-# app/api/v1/models/comments.py
+from sqlalchemy.orm import Session
+from app.api.v1.models.comments import Comments
+from app.api.utils import get_current_timestamp
 
-from sqlalchemy import Column, ForeignKey, Integer, String, Text
+def get_comments(db: Session, id: int):
-from sqlalchemy.orm import relationship
+    return db.query(Comments).filter(Comments.id == id).first()
-from app.api.db.database import Base
 
-class Comments(Base):
+def get_all_comments(db: Session):
-    __tablename__ = 'comments'
+    return db.query(Comments).all()
 
-    id = Column(Integer, primary_key=True, index=True)
+def create_comments(db: Session, data: dict):
-    post_id = Column(Integer, ForeignKey('posts.id'))
+    timestamp = get_current_timestamp()
-    user_id = Column(Integer, ForeignKey('users.id'))
+    new_comment = Comments(**data, created_at=timestamp, updated_at=timestamp)
-    content = Column(Text)
+    db.add(new_comment)
-    created_at = Column(String)
+    db.commit()
-    updated_at = Column(String, nullable=True)
+    db.refresh(new_comment)
+    return new_comment
 
-    post = relationship('Post', back_populates='comments')
+def update_comments(db: Session, id: int, data: dict):
-    user = relationship('User', back_populates='comments')
+    comment = db.query(Comments).filter(Comments.id == id)
+    if not comment.first():
+        return None
+    timestamp = get_current_timestamp()
+    comment.update({**data, "updated_at": timestamp}, synchronize_session=False)
+    db.commit()
+    return comment.first()
 
-    def __repr__(self):
+def delete_comments(db: Session, id: int):
-        return f'Comment(id={self.id}, post_id={self.post_id}, user_id={self.user_id}, content="{self.content[:20]}...", created_at="{self.created_at}", updated_at="{self.updated_at}")'
+    comment = db.query(Comments).filter(Comments.id == id).first()
+    if not comment:
+        return None
+    db.delete(comment)
+    db.commit()
+    return comment

app/api/v1/models/posts.py

@@ -1,16 +1,32 @@
-from sqlalchemy import Column, ForeignKey, Integer, String, Text
+from sqlalchemy.orm import Session
-from sqlalchemy.orm import relationship
+from app.api.v1.models.posts import Posts
-from app.api.db.database import Base
+from app.api.utils import get_current_timestamp
 
-class Posts(Base):
+def get_posts(db: Session, id: int):
-    __tablename__ = 'posts'
+    return db.query(Posts).filter(Posts.id == id).first()
 
-    id = Column(Integer, primary_key=True, index=True)
+def get_all_posts(db: Session):
-    title = Column(String)
+    return db.query(Posts).all()
-    content = Column(Text)
-    user_id = Column(Integer, ForeignKey('users.id'))
 
-    user = relationship('User', back_populates='posts')
+def create_posts(db: Session, data: dict):
+    new_post = Posts(**data, created_at=get_current_timestamp())
+    db.add(new_post)
+    db.commit()
+    db.refresh(new_post)
+    return new_post
 
-    def __repr__(self):
+def update_posts(db: Session, id: int, data: dict):
-        return f'Post(id={self.id}, title={self.title}, content={self.content[:20]}...)'
+    post = db.query(Posts).filter(Posts.id == id)
+    if not post.first():
+        return None
+    post.update(data)
+    db.commit()
+    return post.first()
+
+def delete_posts(db: Session, id: int):
+    post = db.query(Posts).filter(Posts.id == id).first()
+    if not post:
+        return None
+    db.delete(post)
+    db.commit()
+    return post

app/api/v1/models/tags.py

@@ -1,19 +1,31 @@
-# app/api/v1/models/tags.py
+from sqlalchemy.orm import Session
+from app.api.v1.models.tags import Tags
+from app.api.utils import get_current_timestamp
 
-from typing import Optional
+def get_tags(db: Session, id: int):
-from sqlalchemy import Column, ForeignKey, Integer, String, Text
+    return db.query(Tags).filter(Tags.id == id).first()
-from sqlalchemy.orm import relationship
-from app.api.db.database import Base
 
-class Tags(Base):
+def get_all_tags(db: Session):
-    __tablename__ = 'tags'
+    return db.query(Tags).all()
 
-    id = Column(Integer, primary_key=True, index=True)
+def create_tags(db: Session, data: dict):
-    name = Column(String, nullable=False)
+    data['created_at'] = get_current_timestamp()
-    description = Column(Text, nullable=True)
+    new_tag = Tags(**data)
-    post_id = Column(Integer, ForeignKey('posts.id'), nullable=False)
+    db.add(new_tag)
+    db.commit()
+    db.refresh(new_tag)
+    return new_tag
 
-    post = relationship('Post', back_populates='tags')
+def update_tags(db: Session, id: int, data: dict):
+    tag = get_tags(db, id)
+    for key, value in data.items():
+        setattr(tag, key, value)
+    tag.updated_at = get_current_timestamp()
+    db.commit()
+    db.refresh(tag)
+    return tag
 
-    def __repr__(self):
+def delete_tags(db: Session, id: int):
-        return f'Tag(id={self.id}, name={self.name}, description={self.description})'
+    tag = get_tags(db, id)
+    db.delete(tag)
+    db.commit()

app/api/v1/models/user.py

@@ -1,18 +1,32 @@
-from sqlalchemy import Column, ForeignKey, Integer, String, Text
+from sqlalchemy.orm import Session
-from sqlalchemy.orm import relationship
+from app.api.v1.models.user import User
-from app.api.db.database import Base
+from app.api.utils import get_current_timestamp
 
-class User(Base):
+def get_user(db: Session, id: int):
-    __tablename__ = 'user'
+    return db.query(User).filter(User.id == id).first()
 
-    id = Column(Integer, primary_key=True, index=True)
+def get_all_user(db: Session):
-    username = Column(String, unique=True, nullable=False)
+    return db.query(User).all()
-    email = Column(String, unique=True, nullable=False)
-    password_hash = Column(String, nullable=False)
-    is_active = Column(Integer, default=1)
-    bio = Column(Text, nullable=True)
 
-    posts = relationship("Post", back_populates="author")
+def create_user(db: Session, data: dict):
+    new_user = User(**data, created_at=get_current_timestamp(), updated_at=get_current_timestamp())
+    db.add(new_user)
+    db.commit()
+    db.refresh(new_user)
+    return new_user
 
-    def __repr__(self):
+def update_user(db: Session, id: int, data: dict):
-        return f"User(id={self.id}, username='{self.username}', email='{self.email}')"
+    user = db.query(User).filter(User.id == id)
+    if not user.first():
+        return None
+    user.update({**data, "updated_at": get_current_timestamp()})
+    db.commit()
+    return user.first()
+
+def delete_user(db: Session, id: int):
+    user = db.query(User).filter(User.id == id).first()
+    if not user:
+        return None
+    db.delete(user)
+    db.commit()
+    return user

app/api/v1/routes/comments.py

@@ -1,9 +1,39 @@
 from fastapi import APIRouter, Depends, HTTPException
 from sqlalchemy.orm import Session
 from app.api.core.dependencies import get_db
+from app.api.v1.services.comments import get_comments, get_all_comments, create_comments, update_comments, delete_comments
+from app.api.v1.schemas.comments import CommentsBase
 
 router = APIRouter()
 
-@router.get('/comments/')
+@router.get("/comments/", response_model=list[CommentsBase])
 def read_comments(db: Session = Depends(get_db)):
-    return {'message': 'Read comments'}
+    comments = get_all_comments(db)
+    if not comments:
+        raise HTTPException(status_code=404, detail="Comments not found")
+    return comments
+
+@router.get("/comments/{id}", response_model=CommentsBase)
+def read_comment(id: int, db: Session = Depends(get_db)):
+    comment = get_comments(db, id)
+    if not comment:
+        raise HTTPException(status_code=404, detail="Comment not found")
+    return comment
+
+@router.post("/comments/", response_model=CommentsBase)
+def create_comment(comment: CommentsBase, db: Session = Depends(get_db)):
+    return create_comments(db, comment)
+
+@router.put("/comments/{id}", response_model=CommentsBase)
+def update_comment(id: int, comment: CommentsBase, db: Session = Depends(get_db)):
+    updated_comment = update_comments(db, id, comment)
+    if not updated_comment:
+        raise HTTPException(status_code=404, detail="Comment not found")
+    return updated_comment
+
+@router.delete("/comments/{id}", response_model=CommentsBase)
+def delete_comment(id: int, db: Session = Depends(get_db)):
+    deleted_comment = delete_comments(db, id)
+    if not deleted_comment:
+        raise HTTPException(status_code=404, detail="Comment not found")
+    return deleted_comment

app/api/v1/routes/posts.py

@@ -1,9 +1,37 @@
 from fastapi import APIRouter, Depends, HTTPException
 from sqlalchemy.orm import Session
 from app.api.core.dependencies import get_db
+from app.api.v1.services.posts import get_posts, get_all_posts, create_posts, update_posts, delete_posts
+from app.api.v1.schemas.posts import PostsBase
 
 router = APIRouter()
 
-@router.get('/posts/')
+@router.get("/posts/", response_model=list[PostsBase])
 def read_posts(db: Session = Depends(get_db)):
-    return {'message': 'Read posts'}
+    posts = get_all_posts(db)
+    return posts
+
+@router.get("/posts/{id}", response_model=PostsBase)
+def read_post(id: int, db: Session = Depends(get_db)):
+    post = get_posts(db, id)
+    if not post:
+        raise HTTPException(status_code=404, detail="Post not found")
+    return post
+
+@router.post("/posts/", response_model=PostsBase)
+def create_post(post: PostsBase, db: Session = Depends(get_db)):
+    return create_posts(db, post)
+
+@router.put("/posts/{id}", response_model=PostsBase)
+def update_post(id: int, post: PostsBase, db: Session = Depends(get_db)):
+    updated_post = update_posts(db, id, post)
+    if not updated_post:
+        raise HTTPException(status_code=404, detail="Post not found")
+    return updated_post
+
+@router.delete("/posts/{id}", response_model=PostsBase)
+def delete_post(id: int, db: Session = Depends(get_db)):
+    deleted_post = delete_posts(db, id)
+    if not deleted_post:
+        raise HTTPException(status_code=404, detail="Post not found")
+    return deleted_post

app/api/v1/routes/tags.py

@@ -1,9 +1,37 @@
 from fastapi import APIRouter, Depends, HTTPException
 from sqlalchemy.orm import Session
 from app.api.core.dependencies import get_db
+from app.api.v1.services.tags import get_tags, get_all_tags, create_tags, update_tags, delete_tags
+from app.api.v1.schemas.tags import TagsBase
 
 router = APIRouter()
 
-@router.get('/tags/')
+@router.get("/tags/", response_model=list[TagsBase])
 def read_tags(db: Session = Depends(get_db)):
-    return {'message': 'Read tags'}
+    tags = get_all_tags(db)
+    return tags
+
+@router.get("/tags/{id}", response_model=TagsBase)
+def read_tag(id: int, db: Session = Depends(get_db)):
+    tag = get_tags(db, id)
+    if not tag:
+        raise HTTPException(status_code=404, detail="Tag not found")
+    return tag
+
+@router.post("/tags/", response_model=TagsBase)
+def create_tag(tag: TagsBase, db: Session = Depends(get_db)):
+    return create_tags(db, tag)
+
+@router.put("/tags/{id}", response_model=TagsBase)
+def update_tag(id: int, tag: TagsBase, db: Session = Depends(get_db)):
+    updated_tag = update_tags(db, id, tag)
+    if not updated_tag:
+        raise HTTPException(status_code=404, detail="Tag not found")
+    return updated_tag
+
+@router.delete("/tags/{id}", response_model=TagsBase)
+def delete_tag(id: int, db: Session = Depends(get_db)):
+    deleted_tag = delete_tags(db, id)
+    if not deleted_tag:
+        raise HTTPException(status_code=404, detail="Tag not found")
+    return deleted_tag

app/api/v1/routes/user.py

@@ -1,9 +1,37 @@
 from fastapi import APIRouter, Depends, HTTPException
 from sqlalchemy.orm import Session
 from app.api.core.dependencies import get_db
+from app.api.v1.services.user import get_user, get_all_user, create_user, update_user, delete_user
+from app.api.v1.schemas.user import UserBase
 
 router = APIRouter()
 
-@router.get('/user/')
+@router.get("/user/", response_model=list)
-def read_user(db: Session = Depends(get_db)):
+def read_users(db: Session = Depends(get_db)):
-    return {'message': 'Read user'}
+    users = get_all_user(db)
+    return users
+
+@router.get("/user/{id}", response_model=UserBase)
+def read_user(id: int, db: Session = Depends(get_db)):
+    user = get_user(db, id)
+    if not user:
+        raise HTTPException(status_code=404, detail="User not found")
+    return user
+
+@router.post("/user/", response_model=UserBase)
+def create_new_user(user: UserBase, db: Session = Depends(get_db)):
+    return create_user(db, user)
+
+@router.put("/user/{id}", response_model=UserBase)
+def update_user_data(id: int, user: UserBase, db: Session = Depends(get_db)):
+    user_data = get_user(db, id)
+    if not user_data:
+        raise HTTPException(status_code=404, detail="User not found")
+    return update_user(db, user_data, user)
+
+@router.delete("/user/{id}", response_model=UserBase)
+def remove_user(id: int, db: Session = Depends(get_db)):
+    user_data = get_user(db, id)
+    if not user_data:
+        raise HTTPException(status_code=404, detail="User not found")
+    return delete_user(db, user_data)

app/api/v1/schemas/comments.py

@@ -1,10 +1,15 @@
 from pydantic import BaseModel
+from typing import Optional
 
 class CommentsBase(BaseModel):
-    pass
+    body: str
+    post_id: int
+    user_id: int
 
-class Comments(BaseModel):
+class Comments(CommentsBase):
     id: int
+    post_id: int
+    user_id: int
 
     class Config:
         orm_mode = True

app/api/v1/schemas/posts.py

@@ -9,7 +9,7 @@ class PostsBase(BaseModel):
 
 class Posts(PostsBase):
     id: int
-    owner_id: int
+    user_id: int
 
     class Config:
         orm_mode = True

app/api/v1/schemas/tags.py

@@ -1,3 +1,5 @@
+# app/api/v1/schemas/tags.py
+
 from pydantic import BaseModel
 from typing import Optional
 

app/api/v1/schemas/user.py

@@ -1,10 +1,11 @@
 from pydantic import BaseModel
 
 class UserBase(BaseModel):
-    pass
+    title: str
 
 class User(BaseModel):
     id: int
+    title: str
 
     class Config:
         orm_mode = True

app/api/v1/services/comments.py

@@ -0,0 +1,32 @@
+from sqlalchemy.orm import Session
+from app.api.v1.models.comments import Comments
+from app.api.utils import get_current_timestamp
+
+def get_comments(db: Session, id: int):
+    return db.query(Comments).filter(Comments.id == id).first()
+
+def get_all_comments(db: Session):
+    return db.query(Comments).all()
+
+def create_comments(db: Session, data: dict):
+    obj = Comments(**data, created_at=get_current_timestamp())
+    db.add(obj)
+    db.commit()
+    db.refresh(obj)
+    return obj
+
+def update_comments(db: Session, id: int, data: dict):
+    obj = db.query(Comments).filter(Comments.id == id).first()
+    if obj:
+        for key, value in data.items():
+            setattr(obj, key, value)
+        db.commit()
+        db.refresh(obj)
+    return obj
+
+def delete_comments(db: Session, id: int):
+    obj = db.query(Comments).filter(Comments.id == id).first()
+    if obj:
+        db.delete(obj)
+        db.commit()
+    return obj

app/api/v1/services/posts.py

@@ -0,0 +1,32 @@
+from sqlalchemy.orm import Session
+from app.api.v1.models.posts import Posts
+from app.api.utils import get_current_timestamp
+
+def get_posts(db: Session, id: int):
+    return db.query(Posts).filter(Posts.id == id).first()
+
+def get_all_posts(db: Session):
+    return db.query(Posts).all()
+
+def create_posts(db: Session, data: dict):
+    obj = Posts(**data, created_at=get_current_timestamp())
+    db.add(obj)
+    db.commit()
+    db.refresh(obj)
+    return obj
+
+def update_posts(db: Session, id: int, data: dict):
+    obj = db.query(Posts).filter(Posts.id == id).first()
+    if obj:
+        for key, value in data.items():
+            setattr(obj, key, value)
+        db.commit()
+        db.refresh(obj)
+    return obj
+
+def delete_posts(db: Session, id: int):
+    obj = db.query(Posts).filter(Posts.id == id).first()
+    if obj:
+        db.delete(obj)
+        db.commit()
+    return obj

app/api/v1/services/tags.py

@@ -0,0 +1,32 @@
+from sqlalchemy.orm import Session
+from app.api.v1.models.tags import Tags
+from app.api.utils import get_current_timestamp
+
+def get_tags(db: Session, id: int):
+    return db.query(Tags).filter(Tags.id == id).first()
+
+def get_all_tags(db: Session):
+    return db.query(Tags).all()
+
+def create_tags(db: Session, data: dict):
+    obj = Tags(**data, created_at=get_current_timestamp())
+    db.add(obj)
+    db.commit()
+    db.refresh(obj)
+    return obj
+
+def update_tags(db: Session, id: int, data: dict):
+    obj = db.query(Tags).filter(Tags.id == id).first()
+    if obj:
+        for key, value in data.items():
+            setattr(obj, key, value)
+        db.commit()
+        db.refresh(obj)
+    return obj
+
+def delete_tags(db: Session, id: int):
+    obj = db.query(Tags).filter(Tags.id == id).first()
+    if obj:
+        db.delete(obj)
+        db.commit()
+    return obj

app/api/v1/services/user.py

@@ -0,0 +1,32 @@
+from sqlalchemy.orm import Session
+from app.api.v1.models.user import User
+from app.api.utils import get_current_timestamp
+
+def get_user(db: Session, id: int):
+    return db.query(User).filter(User.id == id).first()
+
+def get_all_user(db: Session):
+    return db.query(User).all()
+
+def create_user(db: Session, data: dict):
+    obj = User(**data, created_at=get_current_timestamp())
+    db.add(obj)
+    db.commit()
+    db.refresh(obj)
+    return obj
+
+def update_user(db: Session, id: int, data: dict):
+    obj = db.query(User).filter(User.id == id).first()
+    if obj:
+        for key, value in data.items():
+            setattr(obj, key, value)
+        db.commit()
+        db.refresh(obj)
+    return obj
+
+def delete_user(db: Session, id: int):
+    obj = db.query(User).filter(User.id == id).first()
+    if obj:
+        db.delete(obj)
+        db.commit()
+    return obj

main.py

@@ -1,10 +1,16 @@
-from fastapi import FastAPI
+import uvicorn
+from fastapi import FastAPI, lifespan
 from app.api.db.database import engine, Base
 from app.api.v1.routes import router
 from app.api.core.middleware.activity_tracker import ActivityTrackerMiddleware
-app = FastAPI()
+
+@lifespan
+async def lifespan_event():
+    Base.metadata.create_all(bind=engine)
+
+app = FastAPI(lifespan=lifespan_event)
 app.add_middleware(ActivityTrackerMiddleware)
 app.include_router(router, prefix='/v1')
-@app.on_event('startup')
+
-def startup():
+if __name__ == '__main__':
-    Base.metadata.create_all(bind=engine)
+    uvicorn.run('main:app', port=8000, reload=True)