forked from s3lph/matemat
Merge branch 'dbfacade-staging-merge' into 'master'
Added balance/stock to User/Product and changed the DatabaseFacade accordingly. See merge request s3lph/matemat!18
This commit is contained in:
s3lph
commit
96657c122b
4 changed files with 178 additions and 24 deletions
|
|
@ -75,14 +75,30 @@ class MatematDatabase(object):
|
|||
users: List[User] = []
|
||||
with self.db.transaction(exclusive=False) as c:
|
||||
for row in c.execute('''
|
||||
SELECT user_id, username, email, is_admin, is_member
|
||||
SELECT user_id, username, email, is_admin, is_member, balance
|
||||
FROM users
|
||||
'''):
|
||||
# Decompose each row and put the values into a User object
|
||||
user_id, username, email, is_admin, is_member = row
|
||||
users.append(User(user_id, username, email, is_admin, is_member))
|
||||
user_id, username, email, is_admin, is_member, balance = row
|
||||
users.append(User(user_id, username, balance, email, is_admin, is_member))
|
||||
return users
|
||||
|
||||
def get_user(self, uid: int) -> User:
|
||||
"""
|
||||
Return a user identified by its user ID.
|
||||
:param uid: The user's ID.
|
||||
"""
|
||||
with self.db.transaction(exclusive=False) as c:
|
||||
# Fetch all values to construct the user
|
||||
c.execute('SELECT user_id, username, email, is_admin, is_member, balance FROM users WHERE user_id = ?',
|
||||
[uid])
|
||||
row = c.fetchone()
|
||||
if row is None:
|
||||
raise ValueError(f'No user with user ID {uid} exists.')
|
||||
# Unpack the row and construct the user
|
||||
user_id, username, email, is_admin, is_member, balance = row
|
||||
return User(user_id, username, balance, email, is_admin, is_member)
|
||||
|
||||
def create_user(self,
|
||||
username: str,
|
||||
password: str,
|
||||
|
|
@ -121,7 +137,7 @@ class MatematDatabase(object):
|
|||
# Fetch the new user's rowid.
|
||||
c.execute('SELECT last_insert_rowid()')
|
||||
user_id = int(c.fetchone()[0])
|
||||
return User(user_id, username, email, admin, member)
|
||||
return User(user_id, username, 0, email, admin, member)
|
||||
|
||||
def login(self, username: str, password: Optional[str] = None, touchkey: Optional[str] = None) -> User:
|
||||
"""
|
||||
|
|
@ -138,14 +154,14 @@ class MatematDatabase(object):
|
|||
raise ValueError('Exactly one of password and touchkey must be provided')
|
||||
with self.db.transaction(exclusive=False) as c:
|
||||
c.execute('''
|
||||
SELECT user_id, username, email, password, touchkey, is_admin, is_member
|
||||
SELECT user_id, username, email, password, touchkey, is_admin, is_member, balance
|
||||
FROM users
|
||||
WHERE username = ?
|
||||
''', [username])
|
||||
row = c.fetchone()
|
||||
if row is None:
|
||||
raise AuthenticationError('User does not exist')
|
||||
user_id, username, email, pwhash, tkhash, admin, member = row
|
||||
user_id, username, email, pwhash, tkhash, admin, member, balance = row
|
||||
if password is not None and not compare_digest(crypt.crypt(password, pwhash), pwhash):
|
||||
raise AuthenticationError('Password mismatch')
|
||||
elif touchkey is not None \
|
||||
|
|
@ -154,7 +170,7 @@ class MatematDatabase(object):
|
|||
raise AuthenticationError('Touchkey mismatch')
|
||||
elif touchkey is not None and tkhash is None:
|
||||
raise AuthenticationError('Touchkey not set')
|
||||
return User(user_id, username, email, admin, member)
|
||||
return User(user_id, username, balance, email, admin, member)
|
||||
|
||||
def change_password(self, user: User, oldpass: str, newpass: str, verify_password: bool = True) -> None:
|
||||
"""
|
||||
|
|
@ -216,30 +232,51 @@ class MatematDatabase(object):
|
|||
'tkhash': tkhash
|
||||
})
|
||||
|
||||
def change_user(self, user: User) -> None:
|
||||
def change_user(self, user: User, **kwargs)\
|
||||
-> None:
|
||||
"""
|
||||
Write changes in the User object to the database.
|
||||
:param user: The user object to update in the database.
|
||||
Commit changes to the user in the database. If writing the requested changes succeeded, the values are updated
|
||||
in the provided user object. Otherwise the user object is left untouched. The user to update is identified by
|
||||
the ID field in the provided user object.
|
||||
|
||||
:param user: The user object to update and to identify the requested user by.
|
||||
:param kwargs: The properties to change.
|
||||
:raises DatabaseConsistencyError: If the user represented by the object does not exist.
|
||||
"""
|
||||
# Resolve the values to change
|
||||
name: str = kwargs['name'] if 'name' in kwargs else user.name
|
||||
email: str = kwargs['email'] if 'email' in kwargs else user.email
|
||||
balance: int = kwargs['balance'] if 'balance' in kwargs else user.balance
|
||||
is_admin: bool = kwargs['is_admin'] if 'is_admin' in kwargs else user.is_admin
|
||||
is_member: bool = kwargs['is_member'] if 'is_member' in kwargs else user.is_member
|
||||
with self.db.transaction() as c:
|
||||
c.execute('''
|
||||
UPDATE users SET
|
||||
username = :username,
|
||||
email = :email,
|
||||
balance = :balance,
|
||||
is_admin = :is_admin,
|
||||
is_member = :is_member,
|
||||
lastchange = STRFTIME('%s', 'now')
|
||||
WHERE user_id = :user_id
|
||||
''', {
|
||||
'user_id': user.id,
|
||||
'email': user.email,
|
||||
'is_admin': user.is_admin,
|
||||
'is_member': user.is_member
|
||||
'username': name,
|
||||
'email': email,
|
||||
'balance': balance,
|
||||
'is_admin': is_admin,
|
||||
'is_member': is_member
|
||||
})
|
||||
affected = c.execute('SELECT changes()').fetchone()[0]
|
||||
if affected != 1:
|
||||
raise DatabaseConsistencyError(
|
||||
f'change_user should affect 1 users row, but affected {affected}')
|
||||
# Only update the actual user object after the changes in the database succeeded
|
||||
user.name = name
|
||||
user.email = email
|
||||
user.balance = balance
|
||||
user.is_admin = is_admin
|
||||
user.is_member = is_member
|
||||
|
||||
def delete_user(self, user: User) -> None:
|
||||
"""
|
||||
|
|
@ -265,13 +302,32 @@ class MatematDatabase(object):
|
|||
products: List[Product] = []
|
||||
with self.db.transaction(exclusive=False) as c:
|
||||
for row in c.execute('''
|
||||
SELECT product_id, name, price_member, price_non_member
|
||||
SELECT product_id, name, price_member, price_non_member, stock
|
||||
FROM products
|
||||
'''):
|
||||
product_id, name, price_member, price_external = row
|
||||
products.append(Product(product_id, name, price_member, price_external))
|
||||
product_id, name, price_member, price_external, stock = row
|
||||
products.append(Product(product_id, name, price_member, price_external, stock))
|
||||
return products
|
||||
|
||||
def get_product(self, pid: int) -> Product:
|
||||
"""
|
||||
Return a product identified by its product ID.
|
||||
:param pid: The products's ID.
|
||||
"""
|
||||
with self.db.transaction(exclusive=False) as c:
|
||||
# Fetch all values to construct the product
|
||||
c.execute('''
|
||||
SELECT product_id, name, price_member, price_non_member, stock
|
||||
FROM products
|
||||
WHERE product_id = ?''',
|
||||
[pid])
|
||||
row = c.fetchone()
|
||||
if row is None:
|
||||
raise ValueError(f'No product with product ID {pid} exists.')
|
||||
# Unpack the row and construct the product
|
||||
product_id, name, price_member, price_non_member, stock = row
|
||||
return Product(product_id, name, price_member, price_non_member, stock)
|
||||
|
||||
def create_product(self, name: str, price_member: int, price_non_member: int) -> Product:
|
||||
"""
|
||||
Creates a new product.
|
||||
|
|
@ -296,27 +352,48 @@ class MatematDatabase(object):
|
|||
})
|
||||
c.execute('SELECT last_insert_rowid()')
|
||||
product_id = int(c.fetchone()[0])
|
||||
return Product(product_id, name, price_member, price_non_member)
|
||||
return Product(product_id, name, price_member, price_non_member, 0)
|
||||
|
||||
def change_product(self, product: Product) -> None:
|
||||
def change_product(self, product: Product, **kwargs) -> None:
|
||||
"""
|
||||
Commit changes to the product in the database. If writing the requested changes succeeded, the values are
|
||||
updated in the provided product object. Otherwise the product object is left untouched. The product to update
|
||||
is identified by the ID field in the provided product object.
|
||||
|
||||
:param product: The product object to update and to identify the requested product by.
|
||||
:param kwargs: The properties to change.
|
||||
:raises DatabaseConsistencyError: If the product represented by the object does not exist.
|
||||
"""
|
||||
# Resolve the values to change
|
||||
name: str = kwargs['name'] if 'name' in kwargs else product.name
|
||||
price_member: int = kwargs['price_member'] if 'price_member' in kwargs else product.price_member
|
||||
price_non_member: int = kwargs['price_non_member'] if 'price_non_member' in kwargs else product.price_non_member
|
||||
stock: int = kwargs['stock'] if 'stock' in kwargs else product.stock
|
||||
with self.db.transaction() as c:
|
||||
c.execute('''
|
||||
UPDATE products
|
||||
SET
|
||||
name = :name,
|
||||
price_member = :price_member,
|
||||
price_non_member = :price_non_member
|
||||
price_non_member = :price_non_member,
|
||||
stock = :stock
|
||||
WHERE product_id = :product_id
|
||||
''', {
|
||||
'product_id': product.id,
|
||||
'name': product.name,
|
||||
'price_member': product.price_member,
|
||||
'price_non_member': product.price_non_member
|
||||
'name': name,
|
||||
'price_member': price_member,
|
||||
'price_non_member': price_non_member,
|
||||
'stock': stock
|
||||
})
|
||||
affected = c.execute('SELECT changes()').fetchone()[0]
|
||||
if affected != 1:
|
||||
raise DatabaseConsistencyError(
|
||||
f'change_product should affect 1 products row, but affected {affected}')
|
||||
# Only update the actual product object after the changes in the database succeeded
|
||||
product.name = name
|
||||
product.price_member = price_member
|
||||
product.price_non_member = price_non_member
|
||||
product.stock = stock
|
||||
|
||||
def delete_product(self, product: Product) -> None:
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -26,6 +26,19 @@ class DatabaseTest(unittest.TestCase):
|
|||
with self.assertRaises(ValueError):
|
||||
db.create_user('testuser', 'supersecurepassword2', 'testuser2@example.com')
|
||||
|
||||
def test_get_user(self) -> None:
|
||||
with self.db as db:
|
||||
with db.transaction(exclusive=False):
|
||||
created = db.create_user('testuser', 'supersecurepassword', 'testuser@example.com',
|
||||
admin=True, member=False)
|
||||
user = db.get_user(created.id)
|
||||
self.assertEqual('testuser', user.name)
|
||||
self.assertEqual('testuser@example.com', user.email)
|
||||
self.assertEqual(False, user.is_member)
|
||||
self.assertEqual(True, user.is_admin)
|
||||
with self.assertRaises(ValueError):
|
||||
db.get_user(-1)
|
||||
|
||||
def test_list_users(self) -> None:
|
||||
with self.db as db:
|
||||
users = db.list_users()
|
||||
|
|
@ -170,6 +183,17 @@ class DatabaseTest(unittest.TestCase):
|
|||
with self.assertRaises(ValueError):
|
||||
db.create_product('Club Mate', 250, 250)
|
||||
|
||||
def test_get_product(self) -> None:
|
||||
with self.db as db:
|
||||
with db.transaction(exclusive=False):
|
||||
created = db.create_product('Club Mate', 150, 250)
|
||||
product = db.get_product(created.id)
|
||||
self.assertEqual('Club Mate', product.name)
|
||||
self.assertEqual(150, product.price_member)
|
||||
self.assertEqual(250, product.price_non_member)
|
||||
with self.assertRaises(ValueError):
|
||||
db.get_product(-1)
|
||||
|
||||
def test_list_products(self) -> None:
|
||||
with self.db as db:
|
||||
# Test empty list
|
||||
|
|
|
|||
|
|
@ -3,16 +3,31 @@ from typing import Any
|
|||
|
||||
|
||||
class Product(object):
|
||||
"""
|
||||
Representation of a product offered by the Matemat, with a name, prices for users, and the number of items
|
||||
currently in stock.
|
||||
"""
|
||||
|
||||
def __init__(self,
|
||||
product_id: int,
|
||||
name: str,
|
||||
price_member: int,
|
||||
price_non_member: int) -> None:
|
||||
price_non_member: int,
|
||||
stock: int) -> None:
|
||||
"""
|
||||
Create a new product instance with the given arguments.
|
||||
|
||||
:param product_id: The product ID in the database.
|
||||
:param name: The product's name.
|
||||
:param price_member: The price of a unit of this product for users marked as "members".
|
||||
:param price_non_member: The price of a unit of this product for users NOT marked as "members".
|
||||
:param stock: The number of items of this product currently in stock.
|
||||
"""
|
||||
self._product_id: int = product_id
|
||||
self._name: str = name
|
||||
self._price_member: int = price_member
|
||||
self._price_non_member: int = price_non_member
|
||||
self._stock: int = stock
|
||||
|
||||
def __eq__(self, other: Any) -> bool:
|
||||
if other is None or not isinstance(other, Product):
|
||||
|
|
@ -20,7 +35,8 @@ class Product(object):
|
|||
return self._product_id == other._product_id \
|
||||
and self._name == other._name \
|
||||
and self._price_member == other._price_member \
|
||||
and self._price_non_member == other._price_non_member
|
||||
and self._price_non_member == other._price_non_member \
|
||||
and self._stock == other._stock
|
||||
|
||||
@property
|
||||
def id(self) -> int:
|
||||
|
|
@ -49,3 +65,11 @@ class Product(object):
|
|||
@price_non_member.setter
|
||||
def price_non_member(self, price: int) -> None:
|
||||
self._price_non_member = price
|
||||
|
||||
@property
|
||||
def stock(self) -> int:
|
||||
return self._stock
|
||||
|
||||
@stock.setter
|
||||
def stock(self, stock: int) -> None:
|
||||
self._stock = stock
|
||||
|
|
|
|||
|
|
@ -3,18 +3,35 @@ from typing import Optional, Any
|
|||
|
||||
|
||||
class User(object):
|
||||
"""
|
||||
Representation of a user registered with the Matemat, with a name, e-mail address (optional), whether the user is a
|
||||
member of the organization the Matemat instance is used in, whether the user is an administrator, and the user's
|
||||
account balance.
|
||||
"""
|
||||
|
||||
def __init__(self,
|
||||
user_id: int,
|
||||
username: str,
|
||||
balance: int,
|
||||
email: Optional[str] = None,
|
||||
admin: bool = False,
|
||||
member: bool = True) -> None:
|
||||
"""
|
||||
Create a new user instance with the given arguments.
|
||||
|
||||
:param user_id: The user ID in the database.
|
||||
:param username: The user's name.
|
||||
:param balance: The balance of the user's account.
|
||||
:param email: The user's e-mail address (optional).
|
||||
:param admin: Whether the user is an administrator.
|
||||
:param member: Whether the user is a member.
|
||||
"""
|
||||
self._user_id: int = user_id
|
||||
self._username: str = username
|
||||
self._email: Optional[str] = email
|
||||
self._admin: bool = admin
|
||||
self._member: bool = member
|
||||
self._balance: int = balance
|
||||
|
||||
def __eq__(self, other: Any) -> bool:
|
||||
if other is None or not isinstance(other, User):
|
||||
|
|
@ -33,6 +50,10 @@ class User(object):
|
|||
def name(self) -> str:
|
||||
return self._username
|
||||
|
||||
@name.setter
|
||||
def name(self, value):
|
||||
self._username = value
|
||||
|
||||
@property
|
||||
def email(self) -> Optional[str]:
|
||||
return self._email
|
||||
|
|
@ -56,3 +77,11 @@ class User(object):
|
|||
@is_member.setter
|
||||
def is_member(self, member: bool) -> None:
|
||||
self._member = member
|
||||
|
||||
@property
|
||||
def balance(self) -> int:
|
||||
return self._balance
|
||||
|
||||
@balance.setter
|
||||
def balance(self, balance: int) -> None:
|
||||
self._balance = balance
|
||||
|
|
|
|||
Loading…
Reference in a new issue