87 lines
2.5 KiB
Python
87 lines
2.5 KiB
Python
from email.quoprimime import body_check
|
|
from http.client import HTTPException
|
|
from turtle import title
|
|
from typing import Optional
|
|
from fastapi import FastAPI, Response, status, HTTPException
|
|
from fastapi.params import Body
|
|
from pydantic import BaseModel
|
|
from random import randrange
|
|
import os
|
|
import subprocess
|
|
|
|
app = FastAPI()
|
|
|
|
class Post(BaseModel):
|
|
title: str
|
|
content: str
|
|
published: bool = True
|
|
rating: Optional[int] = None
|
|
|
|
|
|
my_posts= [{"title": "title of post 1", "content": "contents of post 1", "id": 1},{"title": "title of post 2", "content": "contents of post 2", "id": 2}]
|
|
|
|
def find_post(id):
|
|
for p in my_posts:
|
|
if p["id"] == id:
|
|
return p
|
|
|
|
def find_index_post(id):
|
|
for i, p in enumerate(my_posts):
|
|
if p['id'] == id:
|
|
return i
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
return {"message": "Welcome to my api!!"}
|
|
|
|
@app.get("/posts")
|
|
async def get_posts():
|
|
return {"data": my_posts}
|
|
|
|
@app.post("/posts", status_code=status.HTTP_201_CREATED)
|
|
async def create_posts(post: Post):
|
|
post_dict = post.dict()
|
|
post_dict['id'] = randrange(0, 1000000)
|
|
my_posts.append(post_dict)
|
|
return {"data": post_dict}
|
|
|
|
@app.get("/posts/{id}")
|
|
async def get_post(id: int):
|
|
post = find_post(id)
|
|
if not post:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Post id {id} not found")
|
|
return {"post_detail": post}
|
|
|
|
@app.delete("/posts/{id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def delete_post(id: int):
|
|
index = find_index_post(id)
|
|
if index == None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Post id {id} not found")
|
|
my_posts.pop(index)
|
|
|
|
return Response(status_code=status.HTTP_204_NO_CONTENT)
|
|
|
|
@app.put("/posts/{id}")
|
|
async def update_post(id: int, post: Post):
|
|
index = find_index_post(id)
|
|
if index == None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Post id {id} not found")
|
|
post_dict = post.dict()
|
|
post_dict['id'] = id
|
|
my_posts[index] = post_dict
|
|
return {"data": post_dict}
|
|
|
|
|
|
|
|
@app.get("/list")
|
|
async def ping_host():
|
|
# send one packet of data to the host
|
|
# this is specified by '-c 1' in the argument list
|
|
outputlist = []
|
|
# Iterate over all the servers in the list and ping each server
|
|
# get the output as a string
|
|
#output = str(os.system(cmd))
|
|
cmd = subprocess.run(["/bin/ls", "-al"],stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True, text=True)
|
|
# store the output in the list
|
|
output = (f"{cmd.stdout}")
|
|
return output |