-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathmain.py
More file actions
88 lines (65 loc) · 2.03 KB
/
Copy pathmain.py
File metadata and controls
88 lines (65 loc) · 2.03 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
from fastapi import FastAPI
from pydantic import BaseModel
import pymongo
import os
from datetime import datetime
from dotenv import load_dotenv
from bson import ObjectId
This_Api = ""
load_dotenv()
mongo_uri = os.getenv("MONGO_URI")
app = FastAPI()
client = pymongo.MongoClient(mongo_uri)
db = client["blog_platform"]
collection = db["posts"]
class BlogModel(BaseModel):
user_name: str
title: str
description: str
@app.post("/create_blog")
async def create_blog(blog: BlogModel):
current_time = datetime.now()
new_blog = {
"user_name": blog.user_name,
"title": blog.title,
"description": blog.description,
"date": current_time.strftime("%Y-%m-%d"),
"time": current_time.strftime("%H:%M:%S"),
"created_at": current_time,
}
result = collection.insert_one(new_blog)
return {
"message": "Blog created successfully",
"id": This_Api + str(result.inserted_id),
"blog_title": new_blog["title"],
}
@app.get("/get_blog/{blog_id}")
async def get_blog(blog_id: str):
try:
blog = collection.find_one({"_id": ObjectId(blog_id)})
if blog:
blog["_id"] = str(blog["_id"])
return blog
else:
return {"error": "Blog not found"}
except Exception as e:
return {"error": "Invalid ID format"}
# --- 3. Delete Blog (New Route) ---
@app.delete("/delete_blog/{blog_id}")
async def delete_blog(blog_id: str):
try:
result = collection.delete_one({"_id": ObjectId(blog_id)})
if result.deleted_count == 1:
return {"message": "Blog and User data deleted successfully"}
else:
return {"error": "Blog not found with this ID"}
except Exception:
return {"error": "Invalid ID format"}
@app.get("/get_all_blogs")
async def get_all_blogs():
blogs = []
cursor = collection.find({}).sort("created_at", -1)
for blog in cursor:
blog["_id"] = str(blog["_id"])
blogs.append(blog)
return {"total": len(blogs), "data": blogs}