Created
December 8, 2022 01:53
-
-
Save mjtiempo/295837e384c1b4a342cffb49cbe74ca9 to your computer and use it in GitHub Desktop.
fastapi and mongodb crud with file logging
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
''' | |
Generated by ChatGPT using the prompt | |
fastapi crud mongo with logging to file | |
''' | |
from fastapi import FastAPI | |
from pymongo import MongoClient | |
import logging | |
app = FastAPI() | |
# configure logging | |
logging.basicConfig( | |
level=logging.INFO, | |
format="%(asctime)s - %(levelname)s - %(message)s", | |
datefmt="%Y-%m-%d %H:%M:%S", | |
) | |
logger = logging.getLogger(__name__) | |
# create a file handler | |
handler = logging.FileHandler("logs.txt") | |
# add the file handler to the logger | |
logger.addHandler(handler) | |
# connect to MongoDB | |
client = MongoClient("mongodb://localhost:27017/") | |
# create a database | |
db = client["mydatabase"] | |
# create a collection (equivalent to a table in a relational database) | |
collection = db["mycollection"] | |
@app.post("/create") | |
async def create(name: str, age: int): | |
# create a document (equivalent to a row in a relational database) | |
document = {"name": name, "age": age} | |
# insert the document into the collection | |
collection.insert_one(document) | |
# log the operation | |
logger.info(f"Created document: {document}") | |
# return the inserted document | |
return document | |
@app.get("/read") | |
async def read(name: str): | |
# read the document from the collection | |
result = collection.find_one({"name": name}) | |
# log the operation | |
logger.info(f"Read document: {result}") | |
# return the document | |
return result | |
@app.put("/update") | |
async def update(name: str, age: int): | |
# update the document | |
collection.update_one({"name": name}, {"$set": {"age": age}}) | |
# read the updated document | |
result = collection.find_one({"name": name}) | |
# log the operation | |
logger.info(f"Updated document: {result}") | |
# return the updated document | |
return result | |
@app.delete("/delete") | |
async def delete(name: str): | |
# delete the document from the collection | |
collection.delete_one({"name": name}) | |
# log the operation | |
logger.info(f"Deleted document with name: {name}") | |
# return the result of the delete operation | |
return {"status": "success"} |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment