Browse the data

1import asyncio
2import json
3import random
4import string
5from datetime import datetime, timedelta
6from fastapi import FastAPI, HTTPException, Header, Request
7from fastapi.responses import JSONResponse
8from pydantic import BaseModel
9import uvicorn
10 
11app = FastAPI()
12 
13# In-memory storage
14users = {}
15tokens = {}
16invoices = {}
17line_items = {}
18payment_reminders = {}
19webhook_endpoints = {}
20 
21# Simple ID counters
22user_id_counter = 1
23invoice_id_counter = 1
24line_item_id_counter = 1
25reminder_id_counter = 1
26 
27# Models
28class SignupRequest(BaseModel):
29    username: str
30    password: str
31    email: str
32 
33class LoginRequest(BaseModel):
34    username: str
35    password: str
36 
37class InvoiceCreate(BaseModel):
38    client_name: str
39    client_email: str
40    due_date: str
41    tax_rate: float = 0.0
42    notes: str = ""
43 
44class LineItemCreate(BaseModel):
45    invoice_id: int
46    description: str
47    quantity: float
48    unit_price: float
49 
50class WebhookRegister(BaseModel):
51    url: str
52    event: str
53 
54def generate_token():
55    return ''.join(random.choices(string.ascii_letters + string.digits, k=32))
56 
57def get_current_user(authorization: str = Header(None)):
58    if not authorization:
59        raise HTTPException(status_code=401, detail="Missing authorization header")
60    token = authorization.replace("Bearer ", "")
61    if token not in tokens:
62        raise HTTPException(status_code=401, detail="Invalid token")
63    return tokens[token]
64 
65# Auth endpoints
66@app.post("/signup")
67def signup(req: SignupRequest):
68    global user_id_counter
69    for user in users.values():
70        if user["username"] == req.username:
71            raise HTTPException(status_code=400, detail="Username already exists")
72    user_id = user_id_counter
73    user_id_counter += 1
74    users[user_id] = {
75        "id": user_id,
76        "username": req.username,
77        "password": req.password,
78        "email": req.email,
79        "created_at": datetime.now().isoformat()
80    }
81    return {"id": user_id, "username": req.username, "email": req.email}
82 
83@app.post("/login")
84def login(req: LoginRequest):
85    for user in users.values():
86        if user["username"] == req.username and user["password"] == req.password:
87            token = generate_token()
88            tokens[token] = user["id"]
89            return {"token": token, "user_id": user["id"]}
90    raise HTTPException(status_code=401, detail="Invalid credentials")
91 
92# Invoice endpoints
93@app.post("/invoices")
94def create_invoice(inv: InvoiceCreate, authorization: str = Header(None)):
95    user_id = get_current_user(authorization)
96    global invoice_id_counter
97    invoice_id = invoice_id_counter
98    invoice_id_counter += 1
99    invoices[invoice_id] = {
100        "id": invoice_id,
101        "user_id": user_id,
102        "client_name": inv.client_name,
103        "client_email": inv.client_email,
104        "due_date": inv.due_date,
105        "tax_rate": inv.tax_rate,
106        "notes": inv.notes,
107        "created_at": datetime.now().isoformat(),
108        "status": "pending",
109        "line_items": []
110    }
111    return invoices[invoice_id]
112 
113@app.get("/invoices/{invoice_id}")
114def get_invoice(invoice_id: int, authorization: str = Header(None)):
115    user_id = get_current_user(authorization)
116    if invoice_id not in invoices:
117        raise HTTPException(status_code=404, detail="Invoice not found")
118    inv = invoices[invoice_id]
119    if inv["user_id"] != user_id:
120        raise HTTPException(status_code=403, detail="Not your invoice")
121    # Calculate totals
122    subtotal = sum(item["quantity"] * item["unit_price"] for item in inv["line_items"])
123    tax = subtotal * (inv["tax_rate"] / 100)
124    total = subtotal + tax
125    result = inv.copy()
126    result["subtotal"] = subtotal
127    result["tax"] = tax
128    result["total"] = total
129    return result
130 
131@app.post("/invoices/{invoice_id}/send-reminder")
132def send_reminder(invoice_id: int, authorization: str = Header(None)):
133    user_id = get_current_user(authorization)
134    if invoice_id not in invoices:
135        raise HTTPException(status_code=404, detail="Invoice not found")
136    inv = invoices[invoice_id]
137    if inv["user_id"] != user_id:
138        raise HTTPException(status_code=403, detail="Not your invoice")
139    global reminder_id_counter
140    reminder_id = reminder_id_counter
141    reminder_id_counter += 1
142    payment_reminders[reminder_id] = {
143        "id": reminder_id,
144        "invoice_id": invoice_id,
145        "user_id": user_id,
146        "client_email": inv["client_email"],
147        "sent_at": datetime.now().isoformat(),
148        "status": "sent"
149    }
150    # Fire webhooks
151    for wh_id, wh in list(webhook_endpoints.items()):
152        if wh["event"] == "payment_reminder":
153            try:
154                import httpx
155                httpx.post(wh["url"], json={"invoice_id": invoice_id, "client_email": inv["client_email"]})
156            except:
157                pass
158    return payment_reminders[reminder_id]
159 
160# Line item endpoints
161@app.post("/line-items")
162def create_line_item(item: LineItemCreate, authorization: str = Header(None)):
163    user_id = get_current_user(authorization)
164    if item.invoice_id not in invoices:
165        raise HTTPException(status_code=404, detail="Invoice not found")
166    inv = invoices[item.invoice_id]
167    if inv["user_id"] != user_id:
168        raise HTTPException(status_code=403, detail="Not your invoice")
169    global line_item_id_counter
170    li_id = line_item_id_counter
171    line_item_id_counter += 1
172    li = {
173        "id": li_id,
174        "invoice_id": item.invoice_id,
175        "description": item.description,
176        "quantity": item.quantity,
177        "unit_price": item.unit_price,
178        "line_total": item.quantity * item.unit_price
179    }
180    line_items[li_id] = li
181    inv["line_items"].append(li)
182    return li
183 
184@app.get("/line-items/{line_item_id}")
185def get_line_item(line_item_id: int, authorization: str = Header(None)):
186    user_id = get_current_user(authorization)
187    if line_item_id not in line_items:
188        raise HTTPException(status_code=404, detail="Line item not found")
189    li = line_items[line_item_id]
190    inv = invoices.get(li["invoice_id"])
191    if not inv or inv["user_id"] != user_id:
192        raise HTTPException(status_code=403, detail="Not your line item")
193    return li
194 
195# Public invoice view (no auth - for clients)
196@app.get("/public/invoices/{invoice_id}")
197def get_public_invoice(invoice_id: int):
198    if invoice_id not in invoices:
199        raise HTTPException(status_code=404, detail="Invoice not found")
200    inv = invoices[invoice_id]
201    subtotal = sum(item["quantity"] * item["unit_price"] for item in inv["line_items"])
202    tax = subtotal * (inv["tax_rate"] / 100)
203    total = subtotal + tax
204    return {
205        "id": inv["id"],
206        "client_name": inv["client_name"],
207        "client_email": inv["client_email"],
208        "due_date": inv["due_date"],
209        "notes": inv["notes"],
210        "line_items": inv["line_items"],
211        "subtotal": subtotal,
212        "tax": tax,
213        "total": total,
214        "status": inv["status"]
215    }
216 
217# Webhook registration
218@app.post("/webhooks")
219def register_webhook(wh: WebhookRegister, authorization: str = Header(None)):
220    user_id = get_current_user(authorization)
221    wh_id = len(webhook_endpoints) + 1
222    webhook_endpoints[wh_id] = {
223        "id": wh_id,
224        "user_id": user_id,
225        "url": wh.url,
226        "event": wh.event
227    }
228    return webhook_endpoints[wh_id]
229 
230@app.get("/webhooks/{webhook_id}")
231def get_webhook(webhook_id: int, authorization: str = Header(None)):
232    user_id = get_current_user(authorization)
233    if webhook_id not in webhook_endpoints:
234        raise HTTPException(status_code=404, detail="Webhook not found")
235    wh = webhook_endpoints[webhook_id]
236    if wh["user_id"] != user_id:
237        raise HTTPException(status_code=403, detail="Not your webhook")
238    return wh
239 
240@app.get("/users/{user_id}")
241def get_user(user_id: int, authorization: str = Header(None)):
242    current_user_id = get_current_user(authorization)
243    if user_id not in users:
244        raise HTTPException(status_code=404, detail="User not found")
245    if current_user_id != user_id:
246        raise HTTPException(status_code=403, detail="Not your user")
247    return users[user_id]

1fastapi
2uvicorn
3httpx
4pydantic