4fd9fc6221
All checks were successful
Build and Push to Private Registry / build (push) Successful in 9s
113 lines
3.0 KiB
Python
113 lines
3.0 KiB
Python
#!/bin/env python3
|
|
from fastapi import FastAPI, Depends
|
|
import uvicorn
|
|
from sqlalchemy import create_engine, Column, Integer,String, ForeignKey
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.engine import URL
|
|
from sqlalchemy.orm import sessionmaker, Session
|
|
|
|
|
|
|
|
# Database setup
|
|
DATABASE_URL = "sqlite:///./test.db"
|
|
engine = create_engine(DATABASE_URL)
|
|
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
Base = declarative_base()
|
|
|
|
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
class User(Base):
|
|
__tablename__ = "users"
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
name = Column(String, index=True)
|
|
email = Column(String, unique=True, index=True)
|
|
password = Column(String)
|
|
|
|
class Post(Base):
|
|
__tablename__ = "posts"
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
title = Column(String)
|
|
content = Column(String)
|
|
author_id = Column(Integer, ForeignKey("users.id"))
|
|
|
|
|
|
|
|
Base.metadata.create_all(bind=engine)
|
|
app = FastAPI()
|
|
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
return {"message": "Hello World"}
|
|
|
|
@app.get("/users")
|
|
async def get_users(db: Session = Depends(get_db)):
|
|
return db.query(User).all()
|
|
#return [{"id":"asdf1234"}]
|
|
@app.get("/users/{id}")
|
|
async def get_user(id: str):
|
|
return {"id": id}
|
|
|
|
@app.post("/users")
|
|
async def create_user(user: dict,db: Session = Depends(get_db)):
|
|
#db_user = User(name=user["name"], email=user["email"], password=user["password"])
|
|
db_user = User(**user)
|
|
db.add(db_user)
|
|
db.commit()
|
|
db.refresh(db_user)
|
|
return {"message":"User created successfully","user":db_user}
|
|
|
|
@app.get("/posts")
|
|
async def get_posts(db: Session = Depends(get_db)):
|
|
return db.query(Post).all()
|
|
|
|
@app.get("/posts/{id}")
|
|
async def get_post(id: str,db: Session = Depends(get_db)):
|
|
return db.query(Post).filter(Post.id == id).first()
|
|
|
|
@app.post("/posts")
|
|
async def create_post(post: dict,db: Session = Depends(get_db)):
|
|
db_post = Post(**post)
|
|
db.add(db_post)
|
|
db.commit()
|
|
db.refresh(db_post)
|
|
return {"message":"Post created successfully","post":db_post}
|
|
|
|
|
|
@app.put("/posts/{id}")
|
|
async def update_post(id: str, post: dict,db: Session = Depends(get_db)):
|
|
db_post = db.query(Post).filter(Post.id == id).first()
|
|
if "title" in post:
|
|
db_post.title = post["title"]
|
|
if "content" in post:
|
|
db_post.content = post["content"]
|
|
db.commit()
|
|
db.refresh(db_post)
|
|
return {"message":"Post updated successfully","post":db_post}
|
|
|
|
@app.delete("/posts/{id}")
|
|
async def delete_post(id: str, post: dict,db: Session = Depends(get_db)):
|
|
db_post = db.query(Post).filter(Post.id == id).first()
|
|
db.delete(db_post)
|
|
db.commit()
|
|
return {"message":"Post deleted successfully","post":db_post}
|
|
|
|
|
|
@app.get("/hello/{name}")
|
|
async def say_hello(name: str):
|
|
return {"message": f"Hello {name}"}
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
uvicorn.run("papi:app", host="0.0.0.0", port=8912, reload=True, log_level="debug",
|
|
workers=5, limit_concurrency=10, limit_max_requests=100)
|
|
|