Initial commit: locoutil — gestion de location d'outils
This commit is contained in:
@@ -0,0 +1,110 @@
|
||||
import os, uuid, shutil
|
||||
from fastapi import APIRouter, HTTPException, UploadFile, File, Form
|
||||
from pydantic import BaseModel
|
||||
from typing import Optional
|
||||
from ..database import get_db
|
||||
|
||||
router = APIRouter(prefix="/api/clients", tags=["clients"])
|
||||
UPLOAD_DIR = os.environ.get("UPLOAD_DIR", "/uploads")
|
||||
|
||||
class ClientUpdate(BaseModel):
|
||||
name: Optional[str] = None
|
||||
phone: Optional[str] = None
|
||||
email: Optional[str] = None
|
||||
address: Optional[str] = None
|
||||
notes: Optional[str] = None
|
||||
|
||||
def client_full(conn, row):
|
||||
c = dict(row)
|
||||
docs = conn.execute("SELECT * FROM client_documents WHERE client_id = ? ORDER BY id", (c["id"],)).fetchall()
|
||||
c["documents"] = [dict(d) for d in docs]
|
||||
rentals = conn.execute("""
|
||||
SELECT r.*, t.name as tool_name FROM rentals r
|
||||
JOIN tools t ON t.id = r.tool_id
|
||||
WHERE r.client_id = ? ORDER BY r.start_date DESC
|
||||
""", (c["id"],)).fetchall()
|
||||
c["rentals"] = [dict(r) for r in rentals]
|
||||
return c
|
||||
|
||||
@router.get("")
|
||||
def list_clients():
|
||||
with get_db() as conn:
|
||||
rows = conn.execute("SELECT * FROM clients ORDER BY name").fetchall()
|
||||
result = []
|
||||
for r in rows:
|
||||
c = dict(r)
|
||||
c["rental_count"] = conn.execute("SELECT COUNT(*) FROM rentals WHERE client_id = ?", (r["id"],)).fetchone()[0]
|
||||
result.append(c)
|
||||
return result
|
||||
|
||||
@router.get("/{client_id}")
|
||||
def get_client(client_id: int):
|
||||
with get_db() as conn:
|
||||
row = conn.execute("SELECT * FROM clients WHERE id = ?", (client_id,)).fetchone()
|
||||
if not row:
|
||||
raise HTTPException(404, "Client introuvable")
|
||||
return client_full(conn, row)
|
||||
|
||||
@router.post("", status_code=201)
|
||||
def create_client(
|
||||
name: str = Form(...),
|
||||
phone: str = Form(""),
|
||||
email: str = Form(""),
|
||||
address: str = Form(""),
|
||||
notes: str = Form("")
|
||||
):
|
||||
with get_db() as conn:
|
||||
cur = conn.execute(
|
||||
"INSERT INTO clients (name, phone, email, address, notes) VALUES (?,?,?,?,?)",
|
||||
(name.strip(), phone.strip(), email.strip(), address.strip(), notes.strip())
|
||||
)
|
||||
row = conn.execute("SELECT * FROM clients WHERE id = ?", (cur.lastrowid,)).fetchone()
|
||||
return client_full(conn, row)
|
||||
|
||||
@router.put("/{client_id}")
|
||||
def update_client(client_id: int, data: ClientUpdate):
|
||||
with get_db() as conn:
|
||||
if not conn.execute("SELECT id FROM clients WHERE id = ?", (client_id,)).fetchone():
|
||||
raise HTTPException(404, "Client introuvable")
|
||||
fields, values = [], []
|
||||
for field, val in data.model_dump(exclude_none=True).items():
|
||||
fields.append(f"{field} = ?")
|
||||
values.append(val.strip() if isinstance(val, str) else val)
|
||||
if fields:
|
||||
values.append(client_id)
|
||||
conn.execute(f"UPDATE clients SET {', '.join(fields)} WHERE id = ?", values)
|
||||
return client_full(conn, conn.execute("SELECT * FROM clients WHERE id = ?", (client_id,)).fetchone())
|
||||
|
||||
@router.delete("/{client_id}", status_code=204)
|
||||
def delete_client(client_id: int):
|
||||
with get_db() as conn:
|
||||
docs = conn.execute("SELECT filename FROM client_documents WHERE client_id = ?", (client_id,)).fetchall()
|
||||
for doc in docs:
|
||||
path = os.path.join(UPLOAD_DIR, "clients", doc["filename"])
|
||||
if os.path.exists(path):
|
||||
os.remove(path)
|
||||
conn.execute("DELETE FROM clients WHERE id = ?", (client_id,))
|
||||
|
||||
@router.post("/{client_id}/documents", status_code=201)
|
||||
async def add_document(client_id: int, file: UploadFile = File(...), label: str = Form("Document")):
|
||||
with get_db() as conn:
|
||||
if not conn.execute("SELECT id FROM clients WHERE id = ?", (client_id,)).fetchone():
|
||||
raise HTTPException(404, "Client introuvable")
|
||||
ext = os.path.splitext(file.filename)[1].lower()
|
||||
fname = f"{uuid.uuid4().hex}{ext}"
|
||||
path = os.path.join(UPLOAD_DIR, "clients", fname)
|
||||
with open(path, "wb") as f:
|
||||
shutil.copyfileobj(file.file, f)
|
||||
cur = conn.execute("INSERT INTO client_documents (client_id, filename, label) VALUES (?,?,?)", (client_id, fname, label.strip()))
|
||||
return dict(conn.execute("SELECT * FROM client_documents WHERE id = ?", (cur.lastrowid,)).fetchone())
|
||||
|
||||
@router.delete("/{client_id}/documents/{doc_id}", status_code=204)
|
||||
def delete_document(client_id: int, doc_id: int):
|
||||
with get_db() as conn:
|
||||
doc = conn.execute("SELECT * FROM client_documents WHERE id = ? AND client_id = ?", (doc_id, client_id)).fetchone()
|
||||
if not doc:
|
||||
raise HTTPException(404, "Document introuvable")
|
||||
path = os.path.join(UPLOAD_DIR, "clients", doc["filename"])
|
||||
if os.path.exists(path):
|
||||
os.remove(path)
|
||||
conn.execute("DELETE FROM client_documents WHERE id = ?", (doc_id,))
|
||||
@@ -0,0 +1,36 @@
|
||||
from fastapi import APIRouter, HTTPException
|
||||
from pydantic import BaseModel
|
||||
from ..database import get_db
|
||||
|
||||
router = APIRouter(prefix="/api/platforms", tags=["platforms"])
|
||||
|
||||
class PlatformCreate(BaseModel):
|
||||
name: str
|
||||
|
||||
@router.get("")
|
||||
def list_platforms():
|
||||
with get_db() as conn:
|
||||
rows = conn.execute("SELECT * FROM platforms ORDER BY name").fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
@router.post("", status_code=201)
|
||||
def create_platform(data: PlatformCreate):
|
||||
with get_db() as conn:
|
||||
try:
|
||||
cur = conn.execute("INSERT INTO platforms (name) VALUES (?)", (data.name.strip(),))
|
||||
return dict(conn.execute("SELECT * FROM platforms WHERE id = ?", (cur.lastrowid,)).fetchone())
|
||||
except Exception:
|
||||
raise HTTPException(409, "Plateforme déjà existante")
|
||||
|
||||
@router.put("/{platform_id}")
|
||||
def update_platform(platform_id: int, data: PlatformCreate):
|
||||
with get_db() as conn:
|
||||
if not conn.execute("SELECT id FROM platforms WHERE id = ?", (platform_id,)).fetchone():
|
||||
raise HTTPException(404, "Plateforme introuvable")
|
||||
conn.execute("UPDATE platforms SET name = ? WHERE id = ?", (data.name.strip(), platform_id))
|
||||
return dict(conn.execute("SELECT * FROM platforms WHERE id = ?", (platform_id,)).fetchone())
|
||||
|
||||
@router.delete("/{platform_id}", status_code=204)
|
||||
def delete_platform(platform_id: int):
|
||||
with get_db() as conn:
|
||||
conn.execute("DELETE FROM platforms WHERE id = ?", (platform_id,))
|
||||
@@ -0,0 +1,111 @@
|
||||
from fastapi import APIRouter, HTTPException
|
||||
from pydantic import BaseModel
|
||||
from typing import Optional
|
||||
from ..database import get_db
|
||||
|
||||
router = APIRouter(prefix="/api/rentals", tags=["rentals"])
|
||||
|
||||
class RentalCreate(BaseModel):
|
||||
tool_id: int
|
||||
client_id: int
|
||||
platform_id: Optional[int] = None
|
||||
start_date: str
|
||||
end_date: str
|
||||
price: float
|
||||
deposit_collected: bool = False
|
||||
return_notes: str = ""
|
||||
status: str = "confirmed"
|
||||
|
||||
class RentalUpdate(BaseModel):
|
||||
platform_id: Optional[int] = None
|
||||
start_date: Optional[str] = None
|
||||
end_date: Optional[str] = None
|
||||
price: Optional[float] = None
|
||||
deposit_collected: Optional[bool] = None
|
||||
deposit_returned: Optional[bool] = None
|
||||
status: Optional[str] = None
|
||||
return_notes: Optional[str] = None
|
||||
|
||||
def rental_full(conn, row):
|
||||
r = dict(row)
|
||||
tool = conn.execute("SELECT id, name, daily_price, weekend_price, deposit FROM tools WHERE id = ?", (r["tool_id"],)).fetchone()
|
||||
client = conn.execute("SELECT id, name, phone, email FROM clients WHERE id = ?", (r["client_id"],)).fetchone()
|
||||
platform = conn.execute("SELECT id, name FROM platforms WHERE id = ?", (r["platform_id"],)).fetchone() if r.get("platform_id") else None
|
||||
r["tool"] = dict(tool) if tool else None
|
||||
r["client"] = dict(client) if client else None
|
||||
r["platform"] = dict(platform) if platform else None
|
||||
return r
|
||||
|
||||
@router.get("")
|
||||
def list_rentals(status: Optional[str] = None, tool_id: Optional[int] = None, month: Optional[str] = None):
|
||||
with get_db() as conn:
|
||||
q = "SELECT * FROM rentals WHERE 1=1"
|
||||
params = []
|
||||
if status:
|
||||
q += " AND status = ?"; params.append(status)
|
||||
if tool_id:
|
||||
q += " AND tool_id = ?"; params.append(tool_id)
|
||||
if month:
|
||||
q += " AND (start_date LIKE ? OR end_date LIKE ?)"; params += [f"{month}%", f"{month}%"]
|
||||
q += " ORDER BY start_date DESC"
|
||||
rows = conn.execute(q, params).fetchall()
|
||||
return [rental_full(conn, r) for r in rows]
|
||||
|
||||
@router.get("/calendar")
|
||||
def calendar_rentals(year: int, month: int):
|
||||
with get_db() as conn:
|
||||
month_str = f"{year}-{month:02d}"
|
||||
rows = conn.execute("""
|
||||
SELECT r.*, t.name as tool_name, c.name as client_name, p.name as platform_name
|
||||
FROM rentals r
|
||||
JOIN tools t ON t.id = r.tool_id
|
||||
JOIN clients c ON c.id = r.client_id
|
||||
LEFT JOIN platforms p ON p.id = r.platform_id
|
||||
WHERE r.status != 'cancelled'
|
||||
AND r.start_date <= ? AND r.end_date >= ?
|
||||
ORDER BY r.start_date
|
||||
""", (f"{year}-{month:02d}-31", f"{year}-{month:02d}-01")).fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
@router.get("/{rental_id}")
|
||||
def get_rental(rental_id: int):
|
||||
with get_db() as conn:
|
||||
row = conn.execute("SELECT * FROM rentals WHERE id = ?", (rental_id,)).fetchone()
|
||||
if not row:
|
||||
raise HTTPException(404, "Location introuvable")
|
||||
return rental_full(conn, row)
|
||||
|
||||
@router.post("", status_code=201)
|
||||
def create_rental(data: RentalCreate):
|
||||
with get_db() as conn:
|
||||
if not conn.execute("SELECT id FROM tools WHERE id = ?", (data.tool_id,)).fetchone():
|
||||
raise HTTPException(404, "Outil introuvable")
|
||||
if not conn.execute("SELECT id FROM clients WHERE id = ?", (data.client_id,)).fetchone():
|
||||
raise HTTPException(404, "Client introuvable")
|
||||
cur = conn.execute(
|
||||
"INSERT INTO rentals (tool_id, client_id, platform_id, start_date, end_date, price, deposit_collected, status, return_notes) VALUES (?,?,?,?,?,?,?,?,?)",
|
||||
(data.tool_id, data.client_id, data.platform_id, data.start_date, data.end_date, data.price, 1 if data.deposit_collected else 0, data.status, data.return_notes)
|
||||
)
|
||||
return rental_full(conn, conn.execute("SELECT * FROM rentals WHERE id = ?", (cur.lastrowid,)).fetchone())
|
||||
|
||||
@router.put("/{rental_id}")
|
||||
def update_rental(rental_id: int, data: RentalUpdate):
|
||||
with get_db() as conn:
|
||||
if not conn.execute("SELECT id FROM rentals WHERE id = ?", (rental_id,)).fetchone():
|
||||
raise HTTPException(404, "Location introuvable")
|
||||
fields, values = [], []
|
||||
d = data.model_dump(exclude_none=True)
|
||||
for field, val in d.items():
|
||||
if field in ("deposit_collected", "deposit_returned"):
|
||||
fields.append(f"{field} = ?"); values.append(1 if val else 0)
|
||||
else:
|
||||
fields.append(f"{field} = ?"); values.append(val)
|
||||
if fields:
|
||||
values.append(rental_id)
|
||||
conn.execute(f"UPDATE rentals SET {', '.join(fields)} WHERE id = ?", values)
|
||||
return rental_full(conn, conn.execute("SELECT * FROM rentals WHERE id = ?", (rental_id,)).fetchone())
|
||||
|
||||
@router.delete("/{rental_id}", status_code=204)
|
||||
def delete_rental(rental_id: int):
|
||||
with get_db() as conn:
|
||||
conn.execute("DELETE FROM rentals WHERE id = ?", (rental_id,))
|
||||
@@ -0,0 +1,128 @@
|
||||
import os, uuid, shutil
|
||||
from fastapi import APIRouter, HTTPException, UploadFile, File, Form
|
||||
from fastapi.responses import JSONResponse
|
||||
from pydantic import BaseModel
|
||||
from typing import Optional
|
||||
from ..database import get_db
|
||||
|
||||
router = APIRouter(prefix="/api/tools", tags=["tools"])
|
||||
UPLOAD_DIR = os.environ.get("UPLOAD_DIR", "/uploads")
|
||||
|
||||
class ToolUpdate(BaseModel):
|
||||
name: Optional[str] = None
|
||||
description: Optional[str] = None
|
||||
category: Optional[str] = None
|
||||
daily_price: Optional[float] = None
|
||||
weekend_price: Optional[float] = None
|
||||
deposit: Optional[float] = None
|
||||
notes: Optional[str] = None
|
||||
|
||||
def tool_with_images(conn, row):
|
||||
t = dict(row)
|
||||
imgs = conn.execute("SELECT * FROM tool_images WHERE tool_id = ? ORDER BY is_main DESC, id", (t["id"],)).fetchall()
|
||||
t["images"] = [dict(i) for i in imgs]
|
||||
t["main_image"] = next((i["filename"] for i in t["images"] if i["is_main"]), (t["images"][0]["filename"] if t["images"] else None))
|
||||
return t
|
||||
|
||||
@router.get("")
|
||||
def list_tools():
|
||||
with get_db() as conn:
|
||||
rows = conn.execute("SELECT * FROM tools ORDER BY name").fetchall()
|
||||
return [tool_with_images(conn, r) for r in rows]
|
||||
|
||||
@router.get("/{tool_id}")
|
||||
def get_tool(tool_id: int):
|
||||
with get_db() as conn:
|
||||
row = conn.execute("SELECT * FROM tools WHERE id = ?", (tool_id,)).fetchone()
|
||||
if not row:
|
||||
raise HTTPException(404, "Outil introuvable")
|
||||
return tool_with_images(conn, row)
|
||||
|
||||
@router.post("", status_code=201)
|
||||
async def create_tool(
|
||||
name: str = Form(...),
|
||||
description: str = Form(""),
|
||||
category: str = Form(""),
|
||||
daily_price: float = Form(0),
|
||||
weekend_price: float = Form(0),
|
||||
deposit: float = Form(0),
|
||||
notes: str = Form(""),
|
||||
images: list[UploadFile] = File(default=[])
|
||||
):
|
||||
with get_db() as conn:
|
||||
cur = conn.execute(
|
||||
"INSERT INTO tools (name, description, category, daily_price, weekend_price, deposit, notes) VALUES (?,?,?,?,?,?,?)",
|
||||
(name.strip(), description.strip(), category.strip(), daily_price, weekend_price, deposit, notes.strip())
|
||||
)
|
||||
tool_id = cur.lastrowid
|
||||
for i, img in enumerate(images):
|
||||
if img.filename:
|
||||
ext = os.path.splitext(img.filename)[1].lower()
|
||||
fname = f"{uuid.uuid4().hex}{ext}"
|
||||
path = os.path.join(UPLOAD_DIR, "tools", fname)
|
||||
with open(path, "wb") as f:
|
||||
shutil.copyfileobj(img.file, f)
|
||||
conn.execute("INSERT INTO tool_images (tool_id, filename, is_main) VALUES (?,?,?)", (tool_id, fname, 1 if i == 0 else 0))
|
||||
row = conn.execute("SELECT * FROM tools WHERE id = ?", (tool_id,)).fetchone()
|
||||
return tool_with_images(conn, row)
|
||||
|
||||
@router.put("/{tool_id}")
|
||||
def update_tool(tool_id: int, data: ToolUpdate):
|
||||
with get_db() as conn:
|
||||
row = conn.execute("SELECT * FROM tools WHERE id = ?", (tool_id,)).fetchone()
|
||||
if not row:
|
||||
raise HTTPException(404, "Outil introuvable")
|
||||
fields, values = [], []
|
||||
for field, val in data.model_dump(exclude_none=True).items():
|
||||
fields.append(f"{field} = ?")
|
||||
values.append(val.strip() if isinstance(val, str) else val)
|
||||
if fields:
|
||||
values.append(tool_id)
|
||||
conn.execute(f"UPDATE tools SET {', '.join(fields)} WHERE id = ?", values)
|
||||
return tool_with_images(conn, conn.execute("SELECT * FROM tools WHERE id = ?", (tool_id,)).fetchone())
|
||||
|
||||
@router.delete("/{tool_id}", status_code=204)
|
||||
def delete_tool(tool_id: int):
|
||||
with get_db() as conn:
|
||||
imgs = conn.execute("SELECT filename FROM tool_images WHERE tool_id = ?", (tool_id,)).fetchall()
|
||||
for img in imgs:
|
||||
path = os.path.join(UPLOAD_DIR, "tools", img["filename"])
|
||||
if os.path.exists(path):
|
||||
os.remove(path)
|
||||
conn.execute("DELETE FROM tools WHERE id = ?", (tool_id,))
|
||||
|
||||
@router.post("/{tool_id}/images", status_code=201)
|
||||
async def add_image(tool_id: int, image: UploadFile = File(...), is_main: bool = Form(False)):
|
||||
with get_db() as conn:
|
||||
if not conn.execute("SELECT id FROM tools WHERE id = ?", (tool_id,)).fetchone():
|
||||
raise HTTPException(404, "Outil introuvable")
|
||||
ext = os.path.splitext(image.filename)[1].lower()
|
||||
fname = f"{uuid.uuid4().hex}{ext}"
|
||||
path = os.path.join(UPLOAD_DIR, "tools", fname)
|
||||
with open(path, "wb") as f:
|
||||
shutil.copyfileobj(image.file, f)
|
||||
if is_main:
|
||||
conn.execute("UPDATE tool_images SET is_main = 0 WHERE tool_id = ?", (tool_id,))
|
||||
conn.execute("INSERT INTO tool_images (tool_id, filename, is_main) VALUES (?,?,?)", (tool_id, fname, 1 if is_main else 0))
|
||||
return {"filename": fname}
|
||||
|
||||
@router.delete("/{tool_id}/images/{image_id}", status_code=204)
|
||||
def delete_image(tool_id: int, image_id: int):
|
||||
with get_db() as conn:
|
||||
img = conn.execute("SELECT * FROM tool_images WHERE id = ? AND tool_id = ?", (image_id, tool_id)).fetchone()
|
||||
if not img:
|
||||
raise HTTPException(404, "Image introuvable")
|
||||
path = os.path.join(UPLOAD_DIR, "tools", img["filename"])
|
||||
if os.path.exists(path):
|
||||
os.remove(path)
|
||||
conn.execute("DELETE FROM tool_images WHERE id = ?", (image_id,))
|
||||
if img["is_main"]:
|
||||
first = conn.execute("SELECT id FROM tool_images WHERE tool_id = ? LIMIT 1", (tool_id,)).fetchone()
|
||||
if first:
|
||||
conn.execute("UPDATE tool_images SET is_main = 1 WHERE id = ?", (first["id"],))
|
||||
|
||||
@router.post("/{tool_id}/images/{image_id}/main", status_code=204)
|
||||
def set_main_image(tool_id: int, image_id: int):
|
||||
with get_db() as conn:
|
||||
conn.execute("UPDATE tool_images SET is_main = 0 WHERE tool_id = ?", (tool_id,))
|
||||
conn.execute("UPDATE tool_images SET is_main = 1 WHERE id = ? AND tool_id = ?", (image_id, tool_id))
|
||||
Reference in New Issue
Block a user