autocommit 09-07-2024-22-00

This commit is contained in:
Jasen Qin 2024-07-09 22:00:02 +10:00
parent 52808939b7
commit 83c55333c9
8 changed files with 1979 additions and 0 deletions

1
py-kivy/.gitignore vendored Normal file
View File

@ -0,0 +1 @@
.vscode

3
py-kivy/README.md Normal file
View File

@ -0,0 +1,3 @@
# Make poetry stop complaining
This is here to prevent poetry from complaining

124
py-kivy/fastapi_server.py Normal file
View File

@ -0,0 +1,124 @@
import logging
from fastapi import FastAPI, HTTPException
from pymongo import MongoClient
from bson import ObjectId
from pydantic import BaseModel, Field
from typing import List, Optional
# Configure logging
logging.basicConfig(level=logging.DEBUG,
format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
app = FastAPI()
logger.debug("FastAPI application initialized")
# MongoDB connection
client = MongoClient("mongodb://localhost:27017")
db = client["pos_system"]
logger.debug("MongoDB connection established")
# Models
class Item(BaseModel):
name: str
price: float
quantity: int
unit: str
related_items: List[str] = []
class Deal(BaseModel):
name: str
items: List[str]
discount: float
class Order(BaseModel):
customer_name: str
items: List[str]
total_amount: float
payment_method: str
date: str
voucher: Optional[str] = None
logger.debug("Data models defined")
# API Routes
@app.post("/items/")
async def create_item(item: Item):
logger.debug(f"Received request to create item: {item}")
result = db.items.insert_one(item.dict())
logger.debug(f"Item created with ID: {result.inserted_id}")
return {"id": str(result.inserted_id)}
@app.get("/items/")
async def read_items():
logger.debug("Received request to read all items")
items = list(db.items.find())
logger.debug(f"Retrieved {len(items)} items")
return items
@app.get("/items/{item_id}")
async def read_item(item_id: str):
logger.debug(f"Received request to read item with ID: {item_id}")
item = db.items.find_one({"_id": ObjectId(item_id)})
if item:
logger.debug(f"Item found: {item}")
return item
logger.warning(f"Item with ID {item_id} not found")
raise HTTPException(status_code=404, detail="Item not found")
@app.put("/items/{item_id}")
async def update_item(item_id: str, item: Item):
logger.debug(f"Received request to update item with ID: {item_id}")
result = db.items.update_one(
{"_id": ObjectId(item_id)}, {"$set": item.dict()})
if result.modified_count:
logger.debug(f"Item with ID {item_id} updated successfully")
return {"message": "Item updated successfully"}
logger.warning(f"Item with ID {item_id} not found for update")
raise HTTPException(status_code=404, detail="Item not found")
@app.delete("/items/{item_id}")
async def delete_item(item_id: str):
logger.debug(f"Received request to delete item with ID: {item_id}")
result = db.items.delete_one({"_id": ObjectId(item_id)})
if result.deleted_count:
logger.debug(f"Item with ID {item_id} deleted successfully")
return {"message": "Item deleted successfully"}
logger.warning(f"Item with ID {item_id} not found for deletion")
raise HTTPException(status_code=404, detail="Item not found")
# Similar CRUD operations for deals and orders
@app.get("/inventory/")
async def get_inventory():
logger.debug("Received request to get inventory")
inventory = list(db.items.find({}, {"name": 1, "quantity": 1}))
logger.debug(f"Retrieved inventory with {len(inventory)} items")
return inventory
@app.get("/sales/")
async def get_sales_data():
# Implement aggregation for sales data
# This could include total sales, popular items, etc.
pass
if __name__ == "__main__":
logger.info("Starting the FastAPI application")
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
# Add more routes as needed for reporting, analytics, etc.

1595
py-kivy/poetry.lock generated Normal file

File diff suppressed because it is too large Load Diff

46
py-kivy/pos.py Normal file
View File

@ -0,0 +1,46 @@
from kivy.app import App
from kivy.uix.boxlayout import BoxLayout
from kivy.uix.button import Button
from kivy.uix.label import Label
from kivy.uix.scrollview import ScrollView
from kivy.uix.gridlayout import GridLayout
class POSApp(App):
def build(self):
# Main layout
main_layout = BoxLayout(orientation='horizontal')
# Left navigation bar
left_nav = BoxLayout(orientation='vertical', size_hint=(0.2, 1))
left_nav.add_widget(Button(text='Inventory'))
left_nav.add_widget(Button(text='Orders'))
left_nav.add_widget(Button(text='Reports'))
main_layout.add_widget(left_nav)
# Central content area
content_area = ScrollView(size_hint=(0.6, 1))
grid = GridLayout(cols=3, spacing=10, size_hint_y=None)
grid.bind(minimum_height=grid.setter('height'))
# Add some sample content (replace with actual data from API)
for i in range(20):
card = BoxLayout(orientation='vertical', size_hint_y=None, height=200)
card.add_widget(Label(text=f'Item {i}'))
card.add_widget(Label(text=f'Price: ${i*10}'))
card.add_widget(Label(text=f'Stock: {i*5}'))
grid.add_widget(card)
content_area.add_widget(grid)
main_layout.add_widget(content_area)
# Right preview panel (initially hidden)
right_panel = BoxLayout(orientation='vertical', size_hint=(0.2, 1))
right_panel.add_widget(Label(text='Preview Panel'))
main_layout.add_widget(right_panel)
return main_layout
if __name__ == '__main__':
POSApp().run()

22
py-kivy/pyproject.toml Normal file
View File

@ -0,0 +1,22 @@
[tool.poetry]
name = "py-kivy"
version = "0.1.0"
description = ""
authors = ["Jasen Qin <capitalswine@gmail.com>"]
readme = "README.md"
package-mode = false
[tool.poetry.dependencies]
python = "^3.12"
Kivy = "^2.3.0"
fastapi = "^0.111.0"
hypothesis = "^6.105.1"
pymongo = "^4.8.0"
pytest = "^8.2.2"
[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"
[tool.autopep8]
indent-size = 2

43
py-kivy/setup_test_db.py Normal file
View File

@ -0,0 +1,43 @@
# setup_test_db.py
from pymongo import MongoClient
import logging
logging.basicConfig(level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
def setup_test_db():
client = MongoClient("mongodb://localhost:27017")
db = client["test_pos_system"]
# Clear existing data
db.items.drop()
db.orders.drop()
# Insert test items
items = [
{"name": "Apple", "price": 0.5, "quantity": 100, "unit": "piece"},
{"name": "Banana", "price": 0.3, "quantity": 150, "unit": "piece"},
{"name": "Milk", "price": 2.5, "quantity": 50, "unit": "liter"},
{"name": "Bread", "price": 1.5, "quantity": 30, "unit": "loaf"}
]
result = db.items.insert_many(items)
logger.info(f"Inserted {len(result.inserted_ids)} items")
# Insert test orders
orders = [
{"customer_name": "John Doe", "items": [
"Apple", "Milk"], "total_amount": 3.0, "payment_method": "cash", "date": "2024-03-15"},
{"customer_name": "Jane Smith", "items": [
"Banana", "Bread"], "total_amount": 1.8, "payment_method": "credit_card", "date": "2024-03-16"}
]
result = db.orders.insert_many(orders)
logger.info(f"Inserted {len(result.inserted_ids)} orders")
logger.info("Test database setup complete")
if __name__ == "__main__":
setup_test_db()

145
py-kivy/testfastapi.py Normal file
View File

@ -0,0 +1,145 @@
import pytest
from fastapi.testclient import TestClient
from hypothesis import given, strategies as st
from bson import ObjectId
from fastapi_server import app, db # Import your FastAPI app and database connection
client = TestClient(app)
# Helper function to clear the database before each test
@pytest.fixture(autouse=True)
def clear_db():
db.items.delete_many({})
yield
db.items.delete_many({})
# Traditional unit tests
def test_create_item():
response = client.post(
"/items/",
json={"name": "Test Item", "price": 10.99, "quantity": 5, "unit": "piece"}
)
assert response.status_code == 200
assert "id" in response.json()
def test_read_items():
# Create a test item
client.post("/items/", json={"name": "Test Item",
"price": 10.99, "quantity": 5, "unit": "piece"})
response = client.get("/items/")
assert response.status_code == 200
items = response.json()
assert len(items) > 0
assert items[0]["name"] == "Test Item"
def test_read_item():
# Create a test item
create_response = client.post(
"/items/", json={"name": "Test Item", "price": 10.99, "quantity": 5, "unit": "piece"})
item_id = create_response.json()["id"]
response = client.get(f"/items/{item_id}")
assert response.status_code == 200
assert response.json()["name"] == "Test Item"
def test_update_item():
# Create a test item
create_response = client.post(
"/items/", json={"name": "Test Item", "price": 10.99, "quantity": 5, "unit": "piece"})
item_id = create_response.json()["id"]
update_data = {"name": "Updated Item",
"price": 15.99, "quantity": 10, "unit": "piece"}
response = client.put(f"/items/{item_id}", json=update_data)
assert response.status_code == 200
assert response.json()["message"] == "Item updated successfully"
# Verify the update
get_response = client.get(f"/items/{item_id}")
assert get_response.json()["name"] == "Updated Item"
def test_delete_item():
# Create a test item
create_response = client.post(
"/items/", json={"name": "Test Item", "price": 10.99, "quantity": 5, "unit": "piece"})
item_id = create_response.json()["id"]
response = client.delete(f"/items/{item_id}")
assert response.status_code == 200
assert response.json()["message"] == "Item deleted successfully"
# Verify the deletion
get_response = client.get(f"/items/{item_id}")
assert get_response.status_code == 404
# Property-based tests
@given(
name=st.text(min_size=1, max_size=50),
price=st.floats(min_value=0.01, max_value=1000000,
allow_nan=False, allow_infinity=False),
quantity=st.integers(min_value=0, max_value=1000000),
unit=st.text(min_size=1, max_size=10)
)
def test_create_item_property(name, price, quantity, unit):
response = client.post(
"/items/",
json={"name": name, "price": price, "quantity": quantity, "unit": unit}
)
assert response.status_code == 200
assert "id" in response.json()
# Verify the created item
item_id = response.json()["id"]
get_response = client.get(f"/items/{item_id}")
assert get_response.status_code == 200
item = get_response.json()
assert item["name"] == name
assert pytest.approx(item["price"], 0.01) == price
assert item["quantity"] == quantity
assert item["unit"] == unit
@given(
st.lists(
st.fixed_dictionaries({
"name": st.text(min_size=1, max_size=50),
"price": st.floats(min_value=0.01, max_value=1000000, allow_nan=False, allow_infinity=False),
"quantity": st.integers(min_value=0, max_value=1000000),
"unit": st.text(min_size=1, max_size=10)
}),
min_size=1,
max_size=10
)
)
def test_read_items_property(items):
# Create multiple items
for item in items:
client.post("/items/", json=item)
response = client.get("/items/")
assert response.status_code == 200
retrieved_items = response.json()
assert len(retrieved_items) == len(items)
for retrieved_item in retrieved_items:
assert "name" in retrieved_item
assert "price" in retrieved_item
assert "quantity" in retrieved_item
assert "unit" in retrieved_item
# Add more property-based tests for update and delete operations if needed
if __name__ == "__main__":
pytest.main([__file__])