3016498
/
IWS_WS24_clean_architecture
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: 3016498:layer_framework_driver_db
Branches Tags
3016498:layer_interface_adapters
3016498:layer_use_cases
3016498:layer_entities
3016498:main
3016498:layer_framework_driver_db
..
compare: 3016498:main
Branches Tags
3016498:layer_interface_adapters
3016498:layer_use_cases
3016498:layer_entities
3016498:main
3016498:layer_framework_driver_db

1 Commits
layer_fram ... main

Author SHA1 Message Date
michael 78fccb6a94 add .gitignore 2024-11-22 17:59:44 +01:00
7 changed files with 6 additions and 358 deletions
Show Stats Expand all files Collapse all files

4
.gitignore vendored 100644
View File

@ -0,0 +1,4 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class

10
README.md
View File

@ -1,10 +1,4 @@
# IWS_clean_architecture
** Master MDS HSMA **
Clean Architecture - Architekturstil am praktischen Beispiel z.B. in Python.
# Layer: Framework Driver Database
- Implemented a small sqllite3 database for a online shop
** Master MDS HSMA **
Clean Architecture - Architekturstil am praktischen Beispiel z.B. in Python.

94
db/db_cart.py
View File

@ -1,94 +0,0 @@
import sqlite3
class CartDatabase:
def __init__(self, db_file):
"""Initialize the CartDatabase with a connection to the SQLite database."""
self.conn = sqlite3.connect(db_file)
def add_to_cart(self, user_id, product_id, quantity):
"""Add a product to the user's cart."""
sql = "INSERT INTO cart (user_id, product_id, quantity) VALUES (?, ?, ?)"
try:
c = self.conn.cursor()
c.execute(sql, (user_id, product_id, quantity))
self.conn.commit()
return c.lastrowid
except sqlite3.Error as e:
print(e)
def get_cart_for_user(self, user_id):
"""Retrieve the cart items for a specific user."""
sql = "SELECT * FROM cart WHERE user_id = ?"
try:
c = self.conn.cursor()
c.execute(sql, (user_id,))
return c.fetchall()
except sqlite3.Error as e:
print(e)
return []
def update_cart_item(self, user_id, product_id, quantity):
"""Update the quantity of a specific cart item based on user_id and product_id."""
sql = "UPDATE cart SET quantity = ? WHERE user_id = ? AND product_id = ?"
try:
c = self.conn.cursor()
c.execute(sql, (quantity, user_id, product_id))
self.conn.commit()
except sqlite3.Error as e:
print(e)
def remove_from_cart(self, cart_id):
"""Remove a specific item from the cart."""
sql = "DELETE FROM cart WHERE id = ?"
try:
c = self.conn.cursor()
c.execute(sql, (cart_id,))
self.conn.commit()
except sqlite3.Error as e:
print(e)
def clear_cart(self, user_id):
"""Clear all items from a user's cart."""
sql = "DELETE FROM cart WHERE user_id = ?"
try:
c = self.conn.cursor()
c.execute(sql, (user_id,))
self.conn.commit()
except sqlite3.Error as e:
print(e)
def get_cart_total(self, user_id):
"""Calculate the total cost of all items in a user's cart."""
sql = "SELECT SUM(products.price * cart.quantity) FROM cart JOIN products ON cart.product_id = products.id WHERE cart.user_id = ?"
try:
c = self.conn.cursor()
c.execute(sql, (user_id,))
return c.fetchone()[0]
except sqlite3.Error as e:
print(e)
return 0
def main():
database = "db/shop.db"
cart_db = CartDatabase(database)
ex_user_id = 1
# delete all cart items
cart_db.clear_cart(ex_user_id)
# user id, product id, quantity
cart_db.add_to_cart(ex_user_id, 1, 2)
cart_db.add_to_cart(ex_user_id, 3, 5)
print(f"Cart items for user {ex_user_id}:", cart_db.get_cart_for_user(ex_user_id))
print(f"Total cost for user {ex_user_id}:", cart_db.get_cart_total(ex_user_id))
# user id, product id, quantity
cart_db.update_cart_item(ex_user_id, 3, 3)
print(f"Cart items for user {ex_user_id}:", cart_db.get_cart_for_user(ex_user_id))
print(f"Total cost for user {ex_user_id}:", cart_db.get_cart_total(ex_user_id))
if __name__ == "__main__":
main()

101
db/db_product.py
View File

@ -1,101 +0,0 @@
import sqlite3
class ProductDatabase:
def __init__(self, db_file):
"""Initialize the ProductDatabase with a connection to the SQLite database."""
self.conn = sqlite3.connect(db_file)
def add_product(self, product):
"""Add a new product to the products table."""
sql = "INSERT INTO products (name, price, quantity) VALUES (?, ?, ?)"
try:
c = self.conn.cursor()
c.execute(sql, product)
self.conn.commit()
return c.lastrowid
except sqlite3.Error as e:
print(e)
def get_product_by_id(self, product_id):
"""Retrieve a product by its ID."""
sql = "SELECT * FROM products WHERE id = ?"
try:
c = self.conn.cursor()
c.execute(sql, (product_id,))
return c.fetchone()
except sqlite3.Error as e:
print(e)
return None
def get_all_products(self):
"""Retrieve all products."""
sql = "SELECT * FROM products"
try:
c = self.conn.cursor()
c.execute(sql)
return c.fetchall()
except sqlite3.Error as e:
print(e)
return []
def update_product(self, product_id, product_data):
"""Update a product's information."""
sql = "UPDATE products SET name = ?, price = ?, quantity = ? WHERE id = ?"
try:
c = self.conn.cursor()
c.execute(sql, (*product_data, product_id))
self.conn.commit()
except sqlite3.Error as e:
print(e)
def delete_product(self, product_id):
"""Delete a product by its ID."""
sql = "DELETE FROM products WHERE id = ?"
try:
c = self.conn.cursor()
c.execute(sql, (product_id,))
self.conn.commit()
except sqlite3.Error as e:
print(e)
def main():
database = "db/shop.db"
product_db = ProductDatabase(database)
# delete all products
for product in product_db.get_all_products():
product_db.delete_product(product[0])
# name, price, quantity
products = [("Apel", 2.5, 5),
("Banana", 1.5, 10),
("Orange", 3.2, 7)]
for product in products:
product_db.add_product(product)
products = product_db.get_all_products()
print("All products:")
for product in products:
print(product)
# name, price, quantity
product_data = ("Orange", 3.0, 5)
product_db.update_product(3, product_data)
product = product_db.get_product_by_id(1)
print("Updated product:")
print(product)
product_db.delete_product(2)
print("Product deleted")
products = product_db.get_all_products()
print("All products:")
for product in products:
print(product)
if __name__ == "__main__":
main()

66
db/db_setup.py
View File

@ -1,66 +0,0 @@
import sqlite3
def create_connection(db_file):
"""Create a database connection to the SQLite database specified by db_file."""
conn = None
try:
conn = sqlite3.connect(db_file)
print(f"Connected to {db_file}")
except sqlite3.Error as e:
print(e)
return conn
def create_table(conn, create_table_sql):
"""Create a table from the create_table_sql statement."""
try:
c = conn.cursor()
c.execute(create_table_sql)
print("Table created successfully")
except sqlite3.Error as e:
print(e)
def delete_table(conn, table_name):
"""Delete a table specified by table_name."""
try:
c = conn.cursor()
c.execute(f"DROP TABLE IF EXISTS {table_name}")
print(f"Table {table_name} deleted successfully")
except sqlite3.Error as e:
print(e)
def main():
database = "db/shop.db"
sql_user_table = """CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
email TEXT NOT NULL
)"""
sql_product_table = """CREATE TABLE IF NOT EXISTS products (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
price REAL NOT NULL,
quantity INTEGER NOT NULL
)"""
sql_cart_table = """CREATE TABLE IF NOT EXISTS cart (
id INTEGER PRIMARY KEY,
user_id INTEGER NOT NULL,
product_id INTEGER NOT NULL,
quantity INTEGER NOT NULL,
FOREIGN KEY (user_id) REFERENCES users (id),
FOREIGN KEY (product_id) REFERENCES products (id)
)"""
conn = create_connection(database)
if conn is not None:
create_table(conn, sql_user_table)
create_table(conn, sql_product_table)
create_table(conn, sql_cart_table)
else:
print("Error! Cannot create the database connection.")
if __name__ == '__main__':
main()

89
db/db_user.py
View File

@ -1,89 +0,0 @@
import sqlite3
class UserDatabase:
def __init__(self, db_file):
"""Initialize the UserDatabase with a connection to the SQLite database."""
self.conn = sqlite3.connect(db_file)
def add_user(self, user):
"""Add a new user to the users table and return the new user's id."""
sql = "INSERT INTO users (name, email) VALUES (?, ?)"
try:
c = self.conn.cursor()
c.execute(sql, user)
self.conn.commit()
return c.lastrowid
except sqlite3.Error as e:
print(e)
return None
def get_user_by_id(self, user_id):
"""Retrieve a user by their ID."""
sql = "SELECT * FROM users WHERE id = ?"
try:
c = self.conn.cursor()
c.execute(sql, (user_id,))
return c.fetchone()
except sqlite3.Error as e:
print(e)
return None
def update_user(self, user_id, user_data):
"""Update a user's information."""
sql = "UPDATE users SET name = ?, email = ? WHERE id = ?"
try:
c = self.conn.cursor()
c.execute(sql, (*user_data, user_id))
self.conn.commit()
except sqlite3.Error as e:
print(e)
def delete_user(self, user_id):
"""Delete a user by their ID."""
sql = "DELETE FROM users WHERE id = ?"
try:
c = self.conn.cursor()
c.execute(sql, (user_id,))
self.conn.commit()
except sqlite3.Error as e:
print(e)
def get_all_users(self):
"""Retrieve all users."""
sql = "SELECT * FROM users"
try:
c = self.conn.cursor()
c.execute(sql)
return c.fetchall()
except sqlite3.Error as e:
print(e)
return []
def main():
database = "db/shop.db"
user_db = UserDatabase(database)
# delete all users
for user in user_db.get_all_users():
user_db.delete_user(user[0])
users = [("Michael", "michael@mail.com"),
("Klöara", "klara@mail.com"),
("Arman", "arman@mail.com")]
for user in users:
user_id = user_db.add_user(user)
user = user_db.get_user_by_id(1)
print(f"Retrieved user: {user}")
user_db.update_user(2, ("Klara", "klara@mail.com"))
users = user_db.get_all_users()
for user in users:
print(user)
if __name__ == "__main__":
main()

BIN
db/shop.db
View File

Binary file not shown.