Initial commit
This commit is contained in:
@@ -0,0 +1,168 @@
|
||||
import os
|
||||
import sqlite3
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
from flask import Flask, jsonify, render_template, request
|
||||
|
||||
app = Flask(__name__)
|
||||
DATABASE_PATH = os.environ.get("DATABASE_PATH", str(Path(__file__).with_name("inventory.db")))
|
||||
|
||||
|
||||
def get_db():
|
||||
connection = sqlite3.connect(DATABASE_PATH)
|
||||
connection.row_factory = sqlite3.Row
|
||||
connection.execute("PRAGMA foreign_keys = ON")
|
||||
return connection
|
||||
|
||||
|
||||
def initialize_db():
|
||||
Path(DATABASE_PATH).parent.mkdir(parents=True, exist_ok=True)
|
||||
with get_db() as connection:
|
||||
connection.executescript(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS inventory_items (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
name TEXT NOT NULL,
|
||||
item_type TEXT NOT NULL,
|
||||
brand TEXT,
|
||||
model TEXT,
|
||||
serial_number TEXT UNIQUE,
|
||||
asset_tag TEXT UNIQUE,
|
||||
status TEXT NOT NULL DEFAULT 'Available',
|
||||
location TEXT,
|
||||
purchase_date TEXT,
|
||||
notes TEXT,
|
||||
cpu TEXT,
|
||||
ram TEXT,
|
||||
gpu TEXT,
|
||||
storage TEXT,
|
||||
media_format TEXT,
|
||||
created_at TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_inventory_search ON inventory_items(name, brand, model, serial_number, asset_tag);
|
||||
CREATE INDEX IF NOT EXISTS idx_inventory_type ON inventory_items(item_type);
|
||||
CREATE INDEX IF NOT EXISTS idx_inventory_status ON inventory_items(status);
|
||||
"""
|
||||
)
|
||||
|
||||
|
||||
def clean(value):
|
||||
return (value or "").strip()
|
||||
|
||||
|
||||
def item_payload(data):
|
||||
return {
|
||||
"name": clean(data.get("name")),
|
||||
"item_type": clean(data.get("item_type")),
|
||||
"brand": clean(data.get("brand")),
|
||||
"model": clean(data.get("model")),
|
||||
"serial_number": clean(data.get("serial_number")) or None,
|
||||
"asset_tag": clean(data.get("asset_tag")) or None,
|
||||
"status": clean(data.get("status")) or "Available",
|
||||
"location": clean(data.get("location")),
|
||||
"purchase_date": clean(data.get("purchase_date")) or None,
|
||||
"notes": clean(data.get("notes")),
|
||||
"cpu": clean(data.get("cpu")),
|
||||
"ram": clean(data.get("ram")),
|
||||
"gpu": clean(data.get("gpu")),
|
||||
"storage": clean(data.get("storage")),
|
||||
"media_format": clean(data.get("media_format")),
|
||||
}
|
||||
|
||||
|
||||
def serialize(row):
|
||||
return dict(row)
|
||||
|
||||
|
||||
@app.route("/")
|
||||
def index():
|
||||
return render_template("index.html")
|
||||
|
||||
|
||||
@app.get("/api/items")
|
||||
def list_items():
|
||||
query = clean(request.args.get("q"))
|
||||
item_type = clean(request.args.get("type"))
|
||||
status = clean(request.args.get("status"))
|
||||
clauses = []
|
||||
params = []
|
||||
if query:
|
||||
clauses.append("(name LIKE ? OR item_type LIKE ? OR brand LIKE ? OR model LIKE ? OR serial_number LIKE ? OR asset_tag LIKE ? OR location LIKE ?)")
|
||||
params.extend([f"%{query}%"] * 7)
|
||||
if item_type:
|
||||
clauses.append("item_type = ?")
|
||||
params.append(item_type)
|
||||
if status:
|
||||
clauses.append("status = ?")
|
||||
params.append(status)
|
||||
where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
|
||||
with get_db() as connection:
|
||||
rows = connection.execute(f"SELECT * FROM inventory_items {where} ORDER BY updated_at DESC, id DESC", params).fetchall()
|
||||
return jsonify([serialize(row) for row in rows])
|
||||
|
||||
|
||||
@app.get("/api/summary")
|
||||
def summary():
|
||||
with get_db() as connection:
|
||||
total = connection.execute("SELECT COUNT(*) FROM inventory_items").fetchone()[0]
|
||||
available = connection.execute("SELECT COUNT(*) FROM inventory_items WHERE status = 'Available'").fetchone()[0]
|
||||
checked_out = connection.execute("SELECT COUNT(*) FROM inventory_items WHERE status = 'Checked out'").fetchone()[0]
|
||||
maintenance = connection.execute("SELECT COUNT(*) FROM inventory_items WHERE status = 'Maintenance'").fetchone()[0]
|
||||
return jsonify({"total": total, "available": available, "checked_out": checked_out, "maintenance": maintenance})
|
||||
|
||||
|
||||
@app.post("/api/items")
|
||||
def create_item():
|
||||
data = item_payload(request.get_json(silent=True) or {})
|
||||
if not data["name"] or not data["item_type"]:
|
||||
return jsonify({"error": "Name and type are required."}), 400
|
||||
now = datetime.now(timezone.utc).isoformat()
|
||||
try:
|
||||
with get_db() as connection:
|
||||
cursor = connection.execute(
|
||||
"""INSERT INTO inventory_items
|
||||
(name, item_type, brand, model, serial_number, asset_tag, status, location, purchase_date, notes, cpu, ram, gpu, storage, media_format, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
||||
(*data.values(), now, now),
|
||||
)
|
||||
row = connection.execute("SELECT * FROM inventory_items WHERE id = ?", (cursor.lastrowid,)).fetchone()
|
||||
return jsonify(serialize(row)), 201
|
||||
except sqlite3.IntegrityError as error:
|
||||
return jsonify({"error": "Serial number or asset tag already exists."}), 409
|
||||
|
||||
|
||||
@app.put("/api/items/<int:item_id>")
|
||||
def update_item(item_id):
|
||||
data = item_payload(request.get_json(silent=True) or {})
|
||||
if not data["name"] or not data["item_type"]:
|
||||
return jsonify({"error": "Name and type are required."}), 400
|
||||
data["updated_at"] = datetime.now(timezone.utc).isoformat()
|
||||
try:
|
||||
with get_db() as connection:
|
||||
result = connection.execute(
|
||||
"""UPDATE inventory_items SET name=?, item_type=?, brand=?, model=?, serial_number=?, asset_tag=?, status=?, location=?, purchase_date=?, notes=?, cpu=?, ram=?, gpu=?, storage=?, media_format=?, updated_at=? WHERE id=?""",
|
||||
(*data.values(), item_id),
|
||||
)
|
||||
if result.rowcount == 0:
|
||||
return jsonify({"error": "Item not found."}), 404
|
||||
row = connection.execute("SELECT * FROM inventory_items WHERE id = ?", (item_id,)).fetchone()
|
||||
return jsonify(serialize(row))
|
||||
except sqlite3.IntegrityError:
|
||||
return jsonify({"error": "Serial number or asset tag already exists."}), 409
|
||||
|
||||
|
||||
@app.delete("/api/items/<int:item_id>")
|
||||
def delete_item(item_id):
|
||||
with get_db() as connection:
|
||||
result = connection.execute("DELETE FROM inventory_items WHERE id = ?", (item_id,))
|
||||
if result.rowcount == 0:
|
||||
return jsonify({"error": "Item not found."}), 404
|
||||
return jsonify({"deleted": True})
|
||||
|
||||
|
||||
initialize_db()
|
||||
|
||||
if __name__ == "__main__":
|
||||
app.run(host="0.0.0.0", port=5000, debug=False)
|
||||
Reference in New Issue
Block a user