remove implementation details in user_repo
parent
1a4656d451
commit
0f246701cd
|
|
@ -11,83 +11,3 @@ class SQLUserRepository(IUserRepository):
|
||||||
"""Initialize the UserDatabase with a connection to the SQLite database."""
|
"""Initialize the UserDatabase with a connection to the SQLite database."""
|
||||||
self.conn = sqlite3.connect(db_file)
|
self.conn = sqlite3.connect(db_file)
|
||||||
self.conn.row_factory = sqlite3.Row
|
self.conn.row_factory = sqlite3.Row
|
||||||
|
|
||||||
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, password) 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) -> Optional[User]:
|
|
||||||
"""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 = "framework_driver/db/shop.db"
|
|
||||||
|
|
||||||
user_db = SQLUserRepository(database)
|
|
||||||
|
|
||||||
# delete all users
|
|
||||||
for user in user_db.get_all_users():
|
|
||||||
user_db.delete_user(user[0])
|
|
||||||
|
|
||||||
users = [("Reiner", "Reiner.Zufall@mail.com"),
|
|
||||||
("Frank N.", "Frank.N.Stein@mail.com"),
|
|
||||||
("Rainer", "Rainer.Wahnsinn@mail.com"),
|
|
||||||
("Clara ", "Clara.Fall@mail.com")]
|
|
||||||
|
|
||||||
# add users
|
|
||||||
for user in users:
|
|
||||||
user_db.add_user(user)
|
|
||||||
|
|
||||||
users = user_db.get_all_users()
|
|
||||||
for user in users:
|
|
||||||
print(user)
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
main()
|
|
||||||
|
|
@ -1,6 +1,4 @@
|
||||||
from framework_driver.db import db_setup
|
from framework_driver.db import db_setup
|
||||||
from interface_adapters import sql_product_repository
|
from interface_adapters import sql_product_repository
|
||||||
from interface_adapters import sql_user_repository
|
|
||||||
db_setup.main()
|
db_setup.main()
|
||||||
sql_product_repository.main()
|
sql_product_repository.main()
|
||||||
sql_user_repository.main()
|
|
||||||
Loading…
Reference in New Issue