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

This commit is contained in:
Backend IM Bot 2025-03-21 09:33:37 +01:00
parent 64dcbd6144
commit ddfc65091e
18 changed files with 205 additions and 386 deletions

View File

@ -1,32 +1,8 @@
Here's the `dependencies.py` file with the `get_db` function: from sqlalchemy.orm import Session
from app.api.db.database import SessionLocal
def get_db():
from typing import Generator
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.core.config import settings
engine = create_engine(
settings.SQLALCHEMY_DATABASE_URI,
connect_args={"check_same_thread": False},
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_db() -> Generator:
try:
db = SessionLocal() db = SessionLocal()
try:
yield db yield db
finally: finally:
db.close() db.close()
Explanation:
1. We import the necessary modules: `typing` for type hints, `create_engine` and `sessionmaker` from SQLAlchemy for creating the database engine and session factory, and `settings` from `app.core.config` for accessing the database URI.
4. The `get_db` function is a dependency function that returns a SQLAlchemy session. It creates a new session using `SessionLocal()`, yields it to the requesting function, and closes the session after the requesting function has finished executing.

View File

@ -1,58 +1,12 @@
from fastapi import Request, Response import time
from typing import Callable, Awaitable from starlette.middleware.base import BaseHTTPMiddleware
from datetime import datetime from starlette.requests import Request
from starlette.responses import Response
from loguru import logger
async def activity_tracker_middleware(request: Request, call_next: Callable[[Request], Awaitable[Response]]): class ActivityTrackerMiddleware(BaseHTTPMiddleware):
start_time = datetime.now() async def dispatch(self, request: Request, call_next):
start_time = time.time()
response: Response = await call_next(request) response = await call_next(request)
process_time = time.time() - start_time
end_time = datetime.now() logger.info(f'{request.method} {request.url} - Process Time: {process_time:.6f} seconds')
response_time = end_time - start_time
log_activity(request, response, response_time)
return response return response
def log_activity(request: Request, response: Response, response_time: datetime.timedelta):
user_id = get_user_id(request)
endpoint = request.url.path
method = request.method
status_code = response.status_code
log_entry = {
"user_id": user_id,
"endpoint": endpoint,
"method": method,
"status_code": status_code,
"response_time": str(response_time),
"timestamp": datetime.now()
}
save_activity_log(log_entry)
print(log_entry)
def get_user_id(request: Request) -> str:
user_id = request.headers.get("X-User-ID")
return user_id or "anonymous"
def save_activity_log(log_entry: dict):
from app.db.database import SessionLocal
from app.db.models import ActivityLog
db = SessionLocal()
activity_log = ActivityLog(**log_entry)
db.add(activity_log)
db.commit()
db.close()
from app.api.core.middleware.activity_tracker import activity_tracker_middleware
app.add_middleware(activity_tracker_middleware)
Note: This code assumes that you have a properly configured FastAPI project structure with SQLite and SQLAlchemy integration. You may need to adjust the imports and paths based on your project's structure.

View File

@ -1,29 +1,7 @@
Here's the `database.py` file for the `blog_app` FastAPI backend, configured with SQLite using SQLAlchemy:
from sqlalchemy import create_engine from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = 'sqlite:///./blog_app.db'
SQLALCHEMY_DATABASE_URL = "sqlite:///./blog_app.db" engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={'check_same_thread': False})
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base() Base = declarative_base()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
Explanation:
1. We import the necessary modules from SQLAlchemy: `create_engine`, `declarative_base`, and `sessionmaker`.
2. We define the SQLite database URL using `SQLALCHEMY_DATABASE_URL = "sqlite:///./blog_app.db"`. This will create a new SQLite database file named `blog_app.db` in the current directory if it doesn't exist, or use the existing one if it does.
3. We create the SQLAlchemy engine using `create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})`. The `connect_args` parameter is required for SQLite to avoid threading issues.
5. We create a `Base` class using `declarative_base()`. This class will be used as the base class for defining database models.

View File

@ -1,4 +1,4 @@
Here's the `comments.py` file with a SQLAlchemy model for comments: Here's a `comments.py` file with a SQLAlchemy model for comments, following the FastAPI project structure with SQLite and SQLAlchemy:
from sqlalchemy import Column, ForeignKey, Integer, String, Text from sqlalchemy import Column, ForeignKey, Integer, String, Text
@ -11,15 +11,14 @@ class Comment(Base):
id = Column(Integer, primary_key=True, index=True) id = Column(Integer, primary_key=True, index=True)
content = Column(Text, nullable=False) content = Column(Text, nullable=False)
author = Column(String(100), nullable=False) author = Column(String, nullable=False)
post_id = Column(Integer, ForeignKey("posts.id"), nullable=False) post_id = Column(Integer, ForeignKey("posts.id"), nullable=False)
post = relationship("Post", back_populates="comments") post = relationship("Post", back_populates="comments")
def __repr__(self): def __repr__(self):
return f"Comment(id={self.id}, author='{self.author}', content='{self.content[:20]}...')" return f"Comment(id={self.id}, content='{self.content[:20]}...', author='{self.author}', post_id={self.post_id})"
Explanation: Explanation:
5. The `content` column stores the text content of the comment and is required (`nullable=False`).
Note: This code assumes that you have already defined the `Post` model in a separate file, and that the necessary imports and database setup are handled elsewhere in your FastAPI project.

View File

@ -1,5 +1,4 @@
Here's the `posts.py` file for the `blog_app` FastAPI backend, defining a SQLAlchemy model for posts: Here's the `posts.py` file for the `app/api/v1/models/` directory of the `blog_app_igblf` FastAPI backend, defining a SQLAlchemy model for posts:
from sqlalchemy import Column, ForeignKey, Integer, String, Text from sqlalchemy import Column, ForeignKey, Integer, String, Text
from sqlalchemy.orm import relationship from sqlalchemy.orm import relationship
@ -10,15 +9,20 @@ class Post(Base):
__tablename__ = "posts" __tablename__ = "posts"
id = Column(Integer, primary_key=True, index=True) id = Column(Integer, primary_key=True, index=True)
title = Column(String, nullable=False) title = Column(String, index=True)
content = Column(Text, nullable=False) content = Column(Text)
user_id = Column(Integer, ForeignKey("users.id"), nullable=False) user_id = Column(Integer, ForeignKey("users.id"))
author = relationship("User", back_populates="posts") author = relationship("User", back_populates="posts")
def __repr__(self): This file defines a `Post` model that inherits from the `Base` class (assumed to be defined in `app.db`). The `Post` model has the following columns:
return f"Post(id={self.id}, title='{self.title}', content='{self.content[:20]}...')"
This code defines a `Post` model that inherits from the `Base` class provided by SQLAlchemy. The `Post` model has the following fields:
- `user_id`: An integer foreign key referencing the `id` of the `User` model (assuming it exists). The `author` attribute is a relationship with the `User` model (assumed to be defined elsewhere), using the `back_populates` parameter to create a bidirectional relationship. This means that each `Post` instance will have an `author` attribute that references the associated `User` instance, and each `User` instance will have a `posts` attribute that contains a list of associated `Post` instances.
Note that this code assumes the following:
2. The `User` model is defined elsewhere (e.g., `app/api/v1/models/users.py`).
3. The necessary imports for SQLAlchemy are available (e.g., `from sqlalchemy import ...`).
Make sure to import this model in the appropriate places (e.g., `app/main.py`) and create the necessary database tables using SQLAlchemy's database migration tools or by running the appropriate commands.

View File

@ -1,5 +1,7 @@
Here's the `tags.py` file for the `app/api/v1/models/` directory in the `blog_app_igblf` FastAPI backend: Here's the `tags.py` file with a SQLAlchemy model for tags, placed in the `app/api/v1/models/` directory:
from typing import Optional
from sqlalchemy import Column, Integer, String from sqlalchemy import Column, Integer, String
from app.db.base_class import Base from app.db.base_class import Base
@ -7,19 +9,21 @@ class Tag(Base):
__tablename__ = "tags" __tablename__ = "tags"
id = Column(Integer, primary_key=True, index=True) id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True) name = Column(String, nullable=False, unique=True, index=True)
slug = Column(String, index=True, unique=True) description = Column(String, nullable=True)
def __repr__(self): def __repr__(self):
return f"Tag(id={self.id}, name='{self.name}', slug='{self.slug}')" return f"Tag(id={self.id}, name='{self.name}', description='{self.description}')"
Explanation: Explanation:
1. We import the necessary modules from SQLAlchemy: `Column`, `Integer`, and `String`. 1. We import the necessary modules: `typing` for type hints, `sqlalchemy` for defining the database model, and `app.db.base_class` for the base class that inherits from `Base` (assuming you have a `base_class.py` file in the `app/db/` directory).
5. We define three columns for the `Tag` model:
- `id`: An integer primary key column with an index.
- `name`: A string column with an index.
- `slug`: A unique string column with an index.
Note: Make sure to adjust the import paths (`app.db.base_class`) according to your project's structure and naming conventions.
Note: This code assumes that you have already set up the necessary database connection and configured SQLAlchemy in your FastAPI application. If not, you'll need to add the required configuration and import statements accordingly.

View File

@ -1,29 +1,31 @@
Here's the `user.py` file for the `app/api/v1/models/` directory of the `blog_app_igblf` FastAPI backend, defining a SQLAlchemy model for the `User` entity: Sure, here's the `user.py` file for the `app/api/v1/models/` directory of the `blog_app` FastAPI backend project using SQLAlchemy with SQLite:
from sqlalchemy import Column, Integer, String, Boolean from sqlalchemy import Column, Integer, String, Boolean
from sqlalchemy.orm import relationship from app.db import Base
from app.db.base_class import Base
class User(Base): class User(Base):
__tablename__ = "users" __tablename__ = "users"
id = Column(Integer, primary_key=True, index=True) id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True, nullable=False) username = Column(String, unique=True, index=True)
username = Column(String, unique=True, index=True, nullable=False) email = Column(String, unique=True, index=True)
hashed_password = Column(String, nullable=False) hashed_password = Column(String)
is_active = Column(Boolean, default=True) is_active = Column(Boolean, default=True)
is_superuser = Column(Boolean, default=False) is_superuser = Column(Boolean, default=False)
posts = relationship("Post", back_populates="author")
def __repr__(self): def __repr__(self):
return f"User(id={self.id}, email='{self.email}', username='{self.username}')" return f"User(id={self.id}, username='{self.username}', email='{self.email}', is_active={self.is_active}, is_superuser={self.is_superuser})"
This file defines a `User` model with the following fields: Explanation:
- `is_active` (Boolean, default=True)
- `is_superuser` (Boolean, default=False)
The `User` model also has a one-to-many relationship with the `Post` model (not defined here), where each user can have multiple posts.
4. We define the following columns for the `User` model:
- `id`: an integer primary key with an index
- `username`: a string column that is unique and indexed
- `email`: a string column that is unique and indexed
- `hashed_password`: a string column to store the hashed password
- `is_active`: a boolean column to indicate if the user is active or not (default is `True`)
- `is_superuser`: a boolean column to indicate if the user is a superuser or not (default is `False`)

View File

@ -1,30 +1,2 @@
Here's the `__init__.py` file for the `app/api/v1/routes/` directory, which aggregates the routers for posts, comments, tags, and users in the `blog_app_igblf` FastAPI backend:
from fastapi import APIRouter from fastapi import APIRouter
from app.api.v1.routes.posts import router as posts_router
from app.api.v1.routes.comments import router as comments_router
from app.api.v1.routes.tags import router as tags_router
from app.api.v1.routes.users import router as users_router
router = APIRouter() router = APIRouter()
router.include_router(posts_router, prefix="/posts", tags=["posts"])
router.include_router(comments_router, prefix="/comments", tags=["comments"])
router.include_router(tags_router, prefix="/tags", tags=["tags"])
router.include_router(users_router, prefix="/users", tags=["users"])
This `__init__.py` file imports the routers for posts, comments, tags, and users from their respective modules (`app.api.v1.routes.posts`, `app.api.v1.routes.comments`, `app.api.v1.routes.tags`, and `app.api.v1.routes.users`). It then creates an instance of `APIRouter` and includes each of the imported routers with their respective prefixes and tags.
By including all the routers in this `__init__.py` file, you can easily mount the combined router in your FastAPI application. For example, in your `main.py` file, you can import this combined router and include it like this:
from fastapi import FastAPI
from app.api.v1.routes import router as api_router
app = FastAPI()
app.include_router(api_router, prefix="/api/v1")
Note: This code assumes that you have separate modules for each of the routers (`app.api.v1.routes.posts`, `app.api.v1.routes.comments`, `app.api.v1.routes.tags`, and `app.api.v1.routes.users`) and that they have been properly defined and imported.

View File

@ -3,14 +3,15 @@ Here's the `comments.py` file with CRUD endpoints for comments:
from typing import List from typing import List
from fastapi import APIRouter, Depends, HTTPException from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from app.db.database import get_db
from app.models import Comment
from app.schemas import CommentCreate, CommentResponse
from app.db.session import get_db router = APIRouter(
from app.models.comment import Comment prefix="/comments",
from app.schemas.comment import CommentCreate, CommentResponse tags=["Comments"],
router = APIRouter() @router.post("/", response_model=CommentResponse)
@router.post("/comments/", response_model=CommentResponse)
def create_comment(comment: CommentCreate, db: Session = Depends(get_db)): def create_comment(comment: CommentCreate, db: Session = Depends(get_db)):
db_comment = Comment(**comment.dict()) db_comment = Comment(**comment.dict())
db.add(db_comment) db.add(db_comment)
@ -18,19 +19,19 @@ def create_comment(comment: CommentCreate, db: Session = Depends(get_db)):
db.refresh(db_comment) db.refresh(db_comment)
return db_comment return db_comment
@router.get("/comments/", response_model=List[CommentResponse]) @router.get("/", response_model=List[CommentResponse])
def read_comments(db: Session = Depends(get_db)): def get_comments(db: Session = Depends(get_db)):
comments = db.query(Comment).all() comments = db.query(Comment).all()
return comments return comments
@router.get("/comments/{comment_id}", response_model=CommentResponse) @router.get("/{comment_id}", response_model=CommentResponse)
def read_comment(comment_id: int, db: Session = Depends(get_db)): def get_comment(comment_id: int, db: Session = Depends(get_db)):
db_comment = db.query(Comment).filter(Comment.id == comment_id).first() comment = db.query(Comment).filter(Comment.id == comment_id).first()
if not db_comment: if not comment:
raise HTTPException(status_code=404, detail="Comment not found") raise HTTPException(status_code=404, detail="Comment not found")
return db_comment return comment
@router.put("/comments/{comment_id}", response_model=CommentResponse) @router.put("/{comment_id}", response_model=CommentResponse)
def update_comment(comment_id: int, comment: CommentCreate, db: Session = Depends(get_db)): def update_comment(comment_id: int, comment: CommentCreate, db: Session = Depends(get_db)):
db_comment = db.query(Comment).filter(Comment.id == comment_id).first() db_comment = db.query(Comment).filter(Comment.id == comment_id).first()
if not db_comment: if not db_comment:
@ -42,17 +43,13 @@ def update_comment(comment_id: int, comment: CommentCreate, db: Session = Depend
db.refresh(db_comment) db.refresh(db_comment)
return db_comment return db_comment
@router.delete("/comments/{comment_id}", response_model=CommentResponse) @router.delete("/{comment_id}", response_model=CommentResponse)
def delete_comment(comment_id: int, db: Session = Depends(get_db)): def delete_comment(comment_id: int, db: Session = Depends(get_db)):
db_comment = db.query(Comment).filter(Comment.id == comment_id).first() comment = db.query(Comment).filter(Comment.id == comment_id).first()
if not db_comment: if not comment:
raise HTTPException(status_code=404, detail="Comment not found") raise HTTPException(status_code=404, detail="Comment not found")
db.delete(db_comment) db.delete(comment)
db.commit() db.commit()
return db_comment return comment
This file defines the following endpoints: This file defines the following endpoints:
Note: Make sure you have the appropriate models and schemas defined in `app/models/comment.py` and `app/schemas/comment.py`, respectively.

View File

@ -1,18 +1,18 @@
Here's the `posts.py` file with CRUD endpoints for posts: Here's the `posts.py` file with CRUD endpoints for posts:
from typing import List from typing import List
from fastapi import APIRouter, Depends, HTTPException from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from app.db import get_db
from app.models import Post from app.models import Post
from app.schemas import PostCreate, PostResponse from app.schemas import PostCreate, PostResponse
from app.database import get_db
router = APIRouter( router = APIRouter(
prefix="/posts", prefix="/posts",
tags=["posts"], tags=["Posts"],
@router.post("/", response_model=PostResponse, status_code=201) @router.post("/", response_model=PostResponse, status_code=status.HTTP_201_CREATED)
def create_post(post: PostCreate, db: Session = Depends(get_db)): def create_post(post: PostCreate, db: Session = Depends(get_db)):
new_post = Post(**post.dict()) new_post = Post(**post.dict())
db.add(new_post) db.add(new_post)
@ -29,30 +29,31 @@ def get_all_posts(db: Session = Depends(get_db)):
def get_post(post_id: int, db: Session = Depends(get_db)): def get_post(post_id: int, db: Session = Depends(get_db)):
post = db.query(Post).filter(Post.id == post_id).first() post = db.query(Post).filter(Post.id == post_id).first()
if not post: if not post:
raise HTTPException(status_code=404, detail="Post not found") raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Post not found")
return post return post
@router.put("/{post_id}", response_model=PostResponse) @router.put("/{post_id}", response_model=PostResponse)
def update_post(post_id: int, post: PostCreate, db: Session = Depends(get_db)): def update_post(post_id: int, post: PostCreate, db: Session = Depends(get_db)):
db_post = db.query(Post).filter(Post.id == post_id).first() db_post = db.query(Post).filter(Post.id == post_id).first()
if not db_post: if not db_post:
raise HTTPException(status_code=404, detail="Post not found") raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Post not found")
for field, value in post.dict().items(): for key, value in post.dict().items():
setattr(db_post, field, value) setattr(db_post, key, value)
db.commit() db.commit()
db.refresh(db_post) db.refresh(db_post)
return db_post return db_post
@router.delete("/{post_id}", status_code=204) @router.delete("/{post_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_post(post_id: int, db: Session = Depends(get_db)): def delete_post(post_id: int, db: Session = Depends(get_db)):
post = db.query(Post).filter(Post.id == post_id).first() post = db.query(Post).filter(Post.id == post_id).first()
if not post: if not post:
raise HTTPException(status_code=404, detail="Post not found") raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Post not found")
db.delete(post) db.delete(post)
db.commit() db.commit()
return None return
This file defines five routes for CRUD operations on posts: This file defines the following endpoints:
Note that this code assumes the existence of the following components:
Note: Make sure you have the `Post` model, `PostCreate` and `PostResponse` schemas, and the `get_db` dependency defined in your project.

View File

@ -1,11 +1,9 @@
Here's an example `tags.py` file with CRUD endpoints for tags:
from typing import List from typing import List
from fastapi import APIRouter, Depends, HTTPException from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from app.db.session import get_db from app.db.session import get_db
from app.models.tag import Tag from app.models import Tag
from app.schemas.tag import TagCreate, TagUpdate, TagOut from app.schemas.tag import TagCreate, TagUpdate, TagOut
router = APIRouter() router = APIRouter()
@ -52,4 +50,11 @@ def delete_tag(tag_id: int, db: Session = Depends(get_db)):
db.commit() db.commit()
return None return None
This file defines the following endpoints: This code defines a set of CRUD endpoints for managing tags in the `blog_app_igblf` FastAPI application. Here's a breakdown of the different endpoints:
1. `POST /tags`: Creates a new tag. It expects a `TagCreate` Pydantic model in the request body and returns the created `TagOut` model with a status code of 201 (Created).
5. `DELETE /tags/{tag_id}`: Deletes a tag by its ID. If the tag is found, it is deleted from the database and returns a 204 (No Content) status code. If the tag is not found, it raises an HTTP 404 exception.

View File

@ -1,58 +1,61 @@
Here's an example `user.py` file with CRUD endpoints for the `User` model in the `app/api/v1/routes/` directory for the `blog_app` FastAPI backend:
from typing import List from typing import List
from fastapi import APIRouter, Depends, HTTPException from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from app.core import deps from app.core.security import get_current_user
from app.schemas.user import UserCreate, UserUpdate, User from app.db.session import get_db
from app.models.user import User as UserModel from app.models import User
from app.schemas.user import UserCreate, UserRead, UserUpdate
router = APIRouter() router = APIRouter(
prefix="/users",
tags=["users"],
@router.post("/users", response_model=User, status_code=201) @router.post("/", response_model=UserRead)
def create_user(user_in: UserCreate, db: Session = Depends(deps.get_db)): def create_user(user: UserCreate, db: Session = Depends(get_db)):
user = UserModel(**user_in.dict()) db_user = User(**user.dict())
db.add(user) db.add(db_user)
db.commit() db.commit()
db.refresh(user) db.refresh(db_user)
return user return db_user
@router.get("/users", response_model=List[User]) @router.get("/", response_model=List[UserRead])
def read_users(db: Session = Depends(deps.get_db)): def read_users(db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
users = db.query(UserModel).all() users = db.query(User).all()
return users return users
@router.get("/users/{user_id}", response_model=User) @router.get("/{user_id}", response_model=UserRead)
def read_user(user_id: int, db: Session = Depends(deps.get_db)): def read_user(user_id: int, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
user = db.query(UserModel).get(user_id) db_user = db.query(User).filter(User.id == user_id).first()
if not user: if not db_user:
raise HTTPException(status_code=404, detail="User not found") raise HTTPException(status_code=404, detail="User not found")
return user return db_user
@router.put("/users/{user_id}", response_model=User) @router.put("/{user_id}", response_model=UserRead)
def update_user(user_id: int, user_in: UserUpdate, db: Session = Depends(deps.get_db)): def update_user(user_id: int, user: UserUpdate, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
user = db.query(UserModel).get(user_id) db_user = db.query(User).filter(User.id == user_id).first()
if not user: if not db_user:
raise HTTPException(status_code=404, detail="User not found") raise HTTPException(status_code=404, detail="User not found")
update_data = user_in.dict(exclude_unset=True) user_data = user.dict(exclude_unset=True)
for field, value in update_data.items(): for key, value in user_data.items():
setattr(user, field, value) setattr(db_user, key, value)
db.add(user) db.add(db_user)
db.commit() db.commit()
db.refresh(user) db.refresh(db_user)
return user return db_user
@router.delete("/users/{user_id}", status_code=204) @router.delete("/{user_id}", response_model=UserRead)
def delete_user(user_id: int, db: Session = Depends(deps.get_db)): def delete_user(user_id: int, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
user = db.query(UserModel).get(user_id) db_user = db.query(User).filter(User.id == user_id).first()
if not user: if not db_user:
raise HTTPException(status_code=404, detail="User not found") raise HTTPException(status_code=404, detail="User not found")
db.delete(user) db.delete(db_user)
db.commit() db.commit()
return None return db_user
This code defines CRUD (Create, Read, Update, Delete) endpoints for the `User` model using the FastAPI router. Here's a breakdown of the code: This file defines the following endpoints:
Note that this code assumes the existence of the following components: The endpoints use SQLAlchemy models and Pydantic schemas for data validation and serialization. The `get_current_user` dependency is used to ensure that only authenticated users can access certain endpoints (e.g., read all users, read a single user, update a user, and delete a user).
Note that you'll need to define the `User`, `UserCreate`, `UserRead`, and `UserUpdate` models and schemas in separate files (e.g., `app/models.py` and `app/schemas/user.py`). Additionally, you'll need to implement the `get_current_user` function in `app/core/security.py` to handle user authentication and authorization.

View File

@ -1,11 +1,14 @@
Here's the `comments.py` file with Pydantic schemas for comments, to be placed in the `app/api/v1/schemas/` directory: Sure, here's the `comments.py` file for the `app/api/v1/schemas/` directory in the `blog_app_igblf` FastAPI backend:
from typing import Optional
from pydantic import BaseModel
from datetime import datetime from datetime import datetime
from typing import Optional
from pydantic import BaseModel, Field
from app.db.base_class import Base
class CommentBase(BaseModel): class CommentBase(BaseModel):
content: str content: str = Field(..., min_length=1, max_length=1000)
class CommentCreate(CommentBase): class CommentCreate(CommentBase):
pass pass
@ -15,9 +18,9 @@ class CommentUpdate(CommentBase):
class CommentInDBBase(CommentBase): class CommentInDBBase(CommentBase):
id: int id: int
post_id: int
created_at: datetime created_at: datetime
updated_at: Optional[datetime] = None updated_at: Optional[datetime] = None
post_id: int
class Config: class Config:
orm_mode = True orm_mode = True
@ -28,11 +31,13 @@ class Comment(CommentInDBBase):
class CommentInDB(CommentInDBBase): class CommentInDB(CommentInDBBase):
pass pass
Here's a breakdown of the file: This file defines the following Pydantic models for comments:
1. We import the necessary modules: `typing` for type hints, `pydantic` for defining data models, and `datetime` for working with date and time objects.
5. `CommentInDBBase` inherits from `CommentBase` and adds fields for the comment's ID (`id`), the associated post ID (`post_id`), the creation timestamp (`created_at`), and an optional update timestamp (`updated_at`). The `Config` class is used to enable the `orm_mode` for compatibility with SQLAlchemy models.
Note: You may need to adjust the import statements and the `Base` class import based on the actual structure of your `blog_app_igblf` project.

View File

@ -1,37 +1,31 @@
Here's the `posts.py` file with Pydantic schemas for posts, to be placed in the `app/api/v1/schemas/` directory: Here's the `posts.py` file with Pydantic schemas for posts, which can be placed in the `app/api/v1/schemas/` directory:
from datetime import datetime
from typing import Optional from typing import Optional
from pydantic import BaseModel
from datetime import datetime
from pydantic import BaseModel, Field class PostCreate(BaseModel):
title: str
content: str
class PostBase(BaseModel): class PostRead(BaseModel):
title: str = Field(..., min_length=3, max_length=100)
content: str = Field(..., min_length=10)
class PostCreate(PostBase):
pass
class PostUpdate(PostBase):
pass
class PostInDBBase(PostBase):
id: int id: int
title: str
content: str
created_at: datetime created_at: datetime
updated_at: Optional[datetime] = None updated_at: Optional[datetime] = None
class Config: class Config:
orm_mode = True orm_mode = True
class Post(PostInDBBase): class PostUpdate(BaseModel):
pass title: Optional[str] = None
content: Optional[str] = None
class PostInDB(PostInDBBase): Explanation:
pass
This file defines the following Pydantic models: 1. `PostCreate` is a Pydantic model that defines the required fields (`title` and `content`) for creating a new post.
3. `PostUpdate` is a Pydantic model that defines the optional fields (`title` and `content`) for updating an existing post.
These models can be used in the FastAPI routes and services to validate and handle the data related to posts. They ensure data consistency and provide a clear separation between the different operations (create, update, read) on posts. Note: Make sure to import the necessary dependencies (`typing`, `pydantic`, and `datetime`) at the top of the file.
Note: Make sure to import the necessary dependencies (`datetime`, `typing`, and `pydantic`) at the top of the file.

View File

@ -7,32 +7,21 @@ class TagBase(BaseModel):
class TagCreate(TagBase): class TagCreate(TagBase):
pass pass
class TagUpdate(TagBase): class TagUpdate(BaseModel):
name: Optional[str] = None name: Optional[str] = None
class TagInDBBase(TagBase): class TagResponse(TagBase):
id: int id: int
name: str
class Config: class Config:
orm_mode = True orm_mode = True
class Tag(TagInDBBase): This file defines the following Pydantic schemas for tags:
pass
class TagInDB(TagInDBBase):
pass
Explanation:
1. We import the necessary modules: `typing` for type hints and `pydantic` for defining data models.
5. `TagInDBBase` inherits from `TagBase` and adds an `id` field, representing the unique identifier of a tag in the database. It also sets `orm_mode = True` in the `Config` class, which allows Pydantic to read data from ORM objects.
The `orm_mode = True` configuration in the `TagResponse` schema tells Pydantic to read data from an ORM model (SQLAlchemy in this case) or a dict-like object, rather than just a dict.
Note: This file assumes that you have the necessary imports and dependencies set up in your FastAPI project structure, including Pydantic and SQLAlchemy (if you're using it for database operations).
Note: Make sure to import and use these schemas in your FastAPI routes and services as needed.

View File

@ -1,4 +1,4 @@
Sure, here's the `user.py` file for the `app/api/v1/schemas/` directory in the `blog_app_igblf` FastAPI backend project: Here's the `user.py` file that defines Pydantic schemas for the user model in the `app/api/v1/schemas/` directory of the `blog_app_igblf` FastAPI backend:
from typing import Optional from typing import Optional
from pydantic import BaseModel, EmailStr from pydantic import BaseModel, EmailStr
@ -6,7 +6,7 @@ from pydantic import BaseModel, EmailStr
class UserBase(BaseModel): class UserBase(BaseModel):
email: EmailStr email: EmailStr
is_active: Optional[bool] = True is_active: Optional[bool] = True
is_superuser: bool = False is_superuser: Optional[bool] = False
full_name: Optional[str] = None full_name: Optional[str] = None
class UserCreate(UserBase): class UserCreate(UserBase):
@ -15,25 +15,17 @@ class UserCreate(UserBase):
class UserUpdate(UserBase): class UserUpdate(UserBase):
password: Optional[str] = None password: Optional[str] = None
class UserInDBBase(UserBase): class User(UserBase):
id: Optional[int] = None id: int
class Config: class Config:
orm_mode = True orm_mode = True
class User(UserInDBBase): This file defines the following Pydantic schemas:
pass
class UserInDB(UserInDBBase):
hashed_password: str
Here's a breakdown of the different schemas:
Note: Make sure to import the necessary dependencies, such as `pydantic` and `typing`, at the top of the file.
Note: Make sure to import the necessary dependencies (`typing`, `pydantic`) at the top of the file.

62
main.py
View File

@ -1,54 +1,10 @@
Here's a `main.py` file for a FastAPI backend named 'blog_app' with necessary imports and middleware, including SQLite and SQLAlchemy support: from fastapi import FastAPI
from app.api.db.database import engine, Base
import os from app.api.v1.routes import router
from typing import List from app.api.core.middleware.activity_tracker import ActivityTrackerMiddleware
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.ext.declarative import declarative_base
SQLALCHEMY_DATABASE_URL = "sqlite:///./blog_app.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
app = FastAPI() app = FastAPI()
app.add_middleware(ActivityTrackerMiddleware)
origins = [ app.include_router(router, prefix='/v1')
"http://localhost", @app.on_event('startup')
"http://localhost:3000", def startup():
Base.metadata.create_all(bind=engine)
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
from models import Base
Base.metadata.create_all(bind=engine)
from routers import blog, user
app.include_router(blog.router)
app.include_router(user.router)
@app.get("/")
def root():
return {"message": "Welcome to the Blog App!"}
This `main.py` file sets up the necessary components for a FastAPI backend with SQLite and SQLAlchemy support. Here's a breakdown of what's included:
7. Creates database tables if they don't exist by calling `Base.metadata.create_all(bind=engine)`.
8. Imports routers for blog and user functionality (assuming you have separate router files).
10. Defines a root path (`/`) that returns a welcome message.
Note that this `main.py` file assumes you have separate files for models (`models.py`) and routers (`routers/blog.py` and `routers/user.py`). You'll need to create those files and define your models and routes accordingly.

View File

@ -1,17 +1,5 @@
Here's the `requirements.txt` file for the `blog_app` FastAPI backend with the specified dependencies: fastapi
uvicorn
fastapi==0.92.0 sqlalchemy
uvicorn==0.20.0 pydantic
sqlalchemy==1.4.46 loguru
pydantic==1.10.4
loguru==0.6.0
This `requirements.txt` file includes the following dependencies:
- `fastapi==0.92.0`: The FastAPI web framework.
- `uvicorn==0.20.0`: The ASGI server used to run the FastAPI application.
- `sqlalchemy==1.4.46`: The Python SQL toolkit and Object-Relational Mapping (ORM) library.
- `pydantic==1.10.4`: Data validation and settings management using Python type annotations.
- `loguru==0.6.0`: A library for logging in Python that is intended to be a more user-friendly and efficient alternative to the standard logging module.
Note: The specific versions listed in this `requirements.txt` file may need to be updated to the latest versions based on your project requirements and compatibility with other dependencies.