Browse the data

1import os
2import hashlib
3import secrets
4from fastapi import FastAPI, HTTPException, Header
5from fastapi.responses import FileResponse
6from pydantic import BaseModel
7from typing import Optional
8 
9app = FastAPI()
10 
11users = {}
12tokens = {}
13books = {}
14book_counter = 0
15 
16BOOKS_DIR = "/var/books"
17 
18class UserCreate(BaseModel):
19    username: str
20    password: str
21 
22class LoginRequest(BaseModel):
23    username: str
24    password: str
25 
26class BookCreate(BaseModel):
27    isbn: str
28    title: str
29    author: str
30 
31def generate_token():
32    return secrets.token_hex(32)
33 
34def hash_password(password: str):
35    return hashlib.sha256(password.encode()).hexdigest()
36 
37def get_current_user(authorization: Optional[str] = Header(None)):
38    if not authorization:
39        raise HTTPException(status_code=401, detail="Missing auth header")
40    if not authorization.startswith("Bearer "):
41        raise HTTPException(status_code=401, detail="Invalid auth header")
42    token = authorization[7:]
43    if token not in tokens:
44        raise HTTPException(status_code=401, detail="Invalid token")
45    return tokens[token]
46 
47@app.post("/signup")
48def signup(user: UserCreate):
49    if user.username in users:
50        raise HTTPException(status_code=400, detail="User already exists")
51    users[user.username] = hash_password(user.password)
52    return {"message": "User created"}
53 
54@app.post("/login")
55def login(req: LoginRequest):
56    if req.username not in users or users[req.username] != hash_password(req.password):
57        raise HTTPException(status_code=401, detail="Invalid credentials")
58    token = generate_token()
59    tokens[token] = req.username
60    return {"token": token}
61 
62@app.post("/books")
63def create_book(book: BookCreate, authorization: Optional[str] = Header(None)):
64    get_current_user(authorization)
65    global book_counter
66    book_counter += 1
67    books[book_counter] = {
68        "id": book_counter,
69        "isbn": book.isbn,
70        "title": book.title,
71        "author": book.author
72    }
73    return books[book_counter]
74 
75@app.get("/books/{book_id}")
76def get_book(book_id: int, authorization: Optional[str] = Header(None)):
77    get_current_user(authorization)
78    if book_id not in books:
79        raise HTTPException(status_code=404, detail="Book not found")
80    return books[book_id]
81 
82@app.get("/book/{isbn}/files/{file_path:path}")
83def get_book_file(isbn: str, file_path: str, authorization: Optional[str] = Header(None)):
84    get_current_user(authorization)
85    full_path = os.path.join(BOOKS_DIR, isbn, file_path)
86    if not os.path.exists(full_path) or not os.path.isfile(full_path):
87        raise HTTPException(status_code=404, detail="File not found")
88    return FileResponse(full_path)
89 
90@app.get("/search")
91def search_books(author: Optional[str] = None, title: Optional[str] = None, authorization: Optional[str] = Header(None)):
92    get_current_user(authorization)
93    results = []
94    for book_id, book in books.items():
95        if author and author.lower() not in book["author"].lower():
96            continue
97        if title and title.lower() not in book["title"].lower():
98            continue
99        isbn = book["isbn"]
100        book_dir = os.path.join(BOOKS_DIR, isbn)
101        if os.path.isdir(book_dir):
102            files = os.listdir(book_dir)
103            results.append({"isbn": isbn, "title": book["title"], "author": book["author"], "files": files})
104        else:
105            results.append({"isbn": isbn, "title": book["title"], "author": book["author"], "files": []})
106    return results

1fastapi
2uvicorn