thuttu/Docker-Inventory
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add possibility to upload images for inventory units

Browse Source
This commit is contained in:
thuttu
2026-07-29 21:12:28 +03:00
parent 9b717e0f09
commit 709c645af2
5 changed files with 165 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app.py
+149 -11
View File
@@ -1,12 +1,17 @@
import os
import sqlite3
import uuid
from datetime import datetime, timezone
from pathlib import Path
from flask import Flask, jsonify, render_template, request
from flask import Flask, jsonify, render_template, request, send_from_directory
from werkzeug.utils import secure_filename
app = Flask(__name__)
DATABASE_PATH = os.environ.get("DATABASE_PATH", str(Path(__file__).with_name("inventory.db")))
UPLOAD_FOLDER = Path(os.environ.get("UPLOAD_FOLDER", str(Path(DATABASE_PATH).parent / "uploads")))
ALLOWED_IMAGE_EXTENSIONS = {"jpg", "jpeg", "png", "gif", "webp"}
app.config["MAX_CONTENT_LENGTH"] = 10 * 1024 * 1024
def get_db():
@@ -18,6 +23,7 @@ def get_db():
def initialize_db():
Path(DATABASE_PATH).parent.mkdir(parents=True, exist_ok=True)
UPLOAD_FOLDER.mkdir(parents=True, exist_ok=True)
with get_db() as connection:
connection.executescript(
"""
@@ -38,14 +44,33 @@ def initialize_db():
gpu TEXT,
storage TEXT,
media_format TEXT,
image_filename 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);
CREATE TABLE IF NOT EXISTS inventory_images (
id INTEGER PRIMARY KEY AUTOINCREMENT,
item_id INTEGER NOT NULL REFERENCES inventory_items(id) ON DELETE CASCADE,
filename TEXT NOT NULL,
original_name TEXT,
created_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_inventory_images_item ON inventory_images(item_id);
"""
)
columns = {row[1] for row in connection.execute("PRAGMA table_info(inventory_items)")}
if "image_filename" not in columns:
connection.execute("ALTER TABLE inventory_items ADD COLUMN image_filename TEXT")
connection.execute(
"""INSERT INTO inventory_images (item_id, filename, original_name, created_at)
SELECT id, image_filename, image_filename, COALESCE(updated_at, datetime('now'))
FROM inventory_items
WHERE image_filename IS NOT NULL
AND NOT EXISTS (SELECT 1 FROM inventory_images WHERE inventory_images.item_id = inventory_items.id)"""
)
def clean(value):
@@ -72,8 +97,52 @@ def item_payload(data):
}
def serialize(row):
return dict(row)
def serialize(row, connection):
item = dict(row)
item["images"] = [
{"id": image["id"], "filename": image["filename"], "original_name": image["original_name"], "url": f"/media/{image['filename']}"}
for image in connection.execute("SELECT id, filename, original_name FROM inventory_images WHERE item_id = ? ORDER BY id", (row["id"],)).fetchall()
]
item["image_url"] = item["images"][0]["url"] if item["images"] else None
return item
def request_data():
return request.form if request.form else (request.get_json(silent=True) or {})
def save_image(image):
if not image or not image.filename:
return None
original_name = secure_filename(image.filename)
extension = Path(original_name).suffix.lower().lstrip(".")
if extension not in ALLOWED_IMAGE_EXTENSIONS:
raise ValueError("Images must be JPG, PNG, GIF, or WebP files.")
filename = f"{uuid.uuid4().hex}.{extension}"
image.save(UPLOAD_FOLDER / filename)
return filename, original_name
def save_images(files):
saved = []
try:
for image in files:
if image and image.filename:
saved.append(save_image(image))
return saved
except ValueError:
for filename, _ in saved:
delete_image(filename)
raise
def delete_image(filename):
if filename:
(UPLOAD_FOLDER / filename).unlink(missing_ok=True)
def files_from_request():
return request.files.getlist("images") or request.files.getlist("image")
@app.route("/")
@@ -81,6 +150,11 @@ def index():
return render_template("index.html")
@app.get("/media/<path:filename>")
def media(filename):
return send_from_directory(UPLOAD_FOLDER, filename)
@app.get("/api/items")
def list_items():
query = clean(request.args.get("q"))
@@ -100,7 +174,7 @@ def list_items():
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])
return jsonify([serialize(row, connection) for row in rows])
@app.get("/api/summary")
@@ -115,9 +189,16 @@ def summary():
@app.post("/api/items")
def create_item():
data = item_payload(request.get_json(silent=True) or {})
data = item_payload(request_data())
if not data["name"] or not data["item_type"]:
return jsonify({"error": "Name and type are required."}), 400
selected_files = [image for image in files_from_request() if image and image.filename]
if len(selected_files) > 5:
return jsonify({"error": "Each inventory item can have up to 5 pictures."}), 400
try:
images = save_images(selected_files)
except ValueError as error:
return jsonify({"error": str(error)}), 400
now = datetime.now(timezone.utc).isoformat()
try:
with get_db() as connection:
@@ -127,41 +208,98 @@ def create_item():
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(*data.values(), now, now),
)
for filename, original_name in images:
connection.execute("INSERT INTO inventory_images (item_id, filename, original_name, created_at) VALUES (?, ?, ?, ?)", (cursor.lastrowid, filename, original_name, now))
row = connection.execute("SELECT * FROM inventory_items WHERE id = ?", (cursor.lastrowid,)).fetchone()
return jsonify(serialize(row)), 201
except sqlite3.IntegrityError as error:
result = serialize(row, connection)
return jsonify(result), 201
except sqlite3.IntegrityError:
for filename, _ in images:
delete_image(filename)
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 {})
data = item_payload(request_data())
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()
new_images = files_from_request()
saved_images = []
try:
with get_db() as connection:
current = connection.execute("SELECT image_filename FROM inventory_items WHERE id = ?", (item_id,)).fetchone()
if not current:
return jsonify({"error": "Item not found."}), 404
existing_count = connection.execute("SELECT COUNT(*) FROM inventory_images WHERE item_id = ?", (item_id,)).fetchone()[0]
if existing_count + len([image for image in new_images if image and image.filename]) > 5:
return jsonify({"error": "Each inventory item can have up to 5 pictures."}), 400
saved_images = save_images(new_images)
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
for filename, original_name in saved_images:
connection.execute("INSERT INTO inventory_images (item_id, filename, original_name, created_at) VALUES (?, ?, ?, ?)", (item_id, filename, original_name, data["updated_at"]))
row = connection.execute("SELECT * FROM inventory_items WHERE id = ?", (item_id,)).fetchone()
return jsonify(serialize(row))
result = serialize(row, connection)
return jsonify(result)
except ValueError as error:
return jsonify({"error": str(error)}), 400
except sqlite3.IntegrityError:
for filename, _ in saved_images:
delete_image(filename)
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:
images = connection.execute("SELECT filename FROM inventory_images WHERE item_id = ?", (item_id,)).fetchall()
result = connection.execute("DELETE FROM inventory_items WHERE id = ?", (item_id,))
if result.rowcount == 0:
return jsonify({"error": "Item not found."}), 404
for image in images:
delete_image(image["filename"])
return jsonify({"deleted": True})
@app.post("/api/items/<int:item_id>/images")
def add_images(item_id):
files = files_from_request()
with get_db() as connection:
if not connection.execute("SELECT id FROM inventory_items WHERE id = ?", (item_id,)).fetchone():
return jsonify({"error": "Item not found."}), 404
current_count = connection.execute("SELECT COUNT(*) FROM inventory_images WHERE item_id = ?", (item_id,)).fetchone()[0]
selected_files = [image for image in files if image and image.filename]
if current_count + len(selected_files) > 5:
return jsonify({"error": "Each inventory item can have up to 5 pictures."}), 400
try:
saved = save_images(selected_files)
except ValueError as error:
return jsonify({"error": str(error)}), 400
now = datetime.now(timezone.utc).isoformat()
with get_db() as connection:
for filename, original_name in saved:
connection.execute("INSERT INTO inventory_images (item_id, filename, original_name, created_at) VALUES (?, ?, ?, ?)", (item_id, filename, original_name, now))
row = connection.execute("SELECT * FROM inventory_items WHERE id = ?", (item_id,)).fetchone()
return jsonify(serialize(row, connection))
@app.delete("/api/items/<int:item_id>/images/<int:image_id>")
def remove_image(item_id, image_id):
with get_db() as connection:
image = connection.execute("SELECT filename FROM inventory_images WHERE id = ? AND item_id = ?", (image_id, item_id)).fetchone()
if not image:
return jsonify({"error": "Picture not found."}), 404
connection.execute("DELETE FROM inventory_images WHERE id = ?", (image_id,))
item = connection.execute("SELECT * FROM inventory_items WHERE id = ?", (item_id,)).fetchone()
result = serialize(item, connection)
delete_image(image["filename"])
return jsonify(result)
initialize_db()
if __name__ == "__main__":