| | from fastapi import FastAPI |
| | from typing import List, Dict, Any |
| | from fastapi.middleware.cors import CORSMiddleware |
| |
|
| | app = FastAPI() |
| | |
| | app.add_middleware( |
| | CORSMiddleware, |
| | allow_origins=["*"], |
| | allow_credentials=True, |
| | allow_methods=["*"], |
| | allow_headers=["*"], |
| | ) |
| |
|
| | class Database: |
| | def __init__(self): |
| | self.store = {} |
| |
|
| | def insert(self, collection_name: str, document: Dict[str, Any]): |
| | if collection_name not in self.store: |
| | self.store[collection_name] = [] |
| | |
| | self.store[collection_name].append(document) |
| | return document |
| |
|
| | def find(self, collection_name: str, query: Dict[str, Any] = None): |
| | if collection_name not in self.store: |
| | return [] |
| |
|
| | if not query: |
| | return self.store[collection_name] |
| | |
| | return [doc for doc in self.store[collection_name] if all(item in doc.items() for item in query.items())] |
| |
|
| | db = Database() |
| |
|
| | @app.get("/") |
| | def home(): |
| | home = "Hello, Welcome to LinDB!" |
| | return home |
| |
|
| | @app.get("/items", response_model=List[Dict[str, Any]]) |
| | async def get_items(): |
| | return db.find('items') |
| |
|
| | @app.post("/items", response_model=Dict[str, Any]) |
| | async def add_item(item: Dict[str, Any]): |
| | return db.insert('items', item) |
| |
|
| | if __name__ == "__main__": |
| | import uvicorn |
| | uvicorn.run(app, host="0.0.0.0", port=3000) |
| |
|