59 lines
2.0 KiB
Python
59 lines
2.0 KiB
Python
from custom_decorators import singleton
|
|
from database import Database
|
|
from loguru import logger
|
|
from utils.datetime import current_timestamp
|
|
|
|
|
|
@singleton
|
|
class OrderRepository:
|
|
def __init__(self, config):
|
|
self.db = Database(config['MYSQL'])
|
|
|
|
def create(self, order):
|
|
timestamp = current_timestamp()
|
|
order.create_timestamp = timestamp
|
|
order.update_timestamp = timestamp
|
|
sql, params = order.insert_sql()
|
|
logger.debug(sql)
|
|
logger.debug(params)
|
|
try:
|
|
cursor = self.db.execute_query(sql, params)
|
|
cursor.close()
|
|
self.db.commit()
|
|
return timestamp
|
|
except Exception:
|
|
self.db.rollback()
|
|
raise
|
|
|
|
def update_status(self, order_id, status):
|
|
try:
|
|
self.db.execute_query("UPDATE orders "
|
|
"SET status = %s, update_timestamp = %s "
|
|
"WHERE order_id = %s",
|
|
[status, current_timestamp(), order_id])
|
|
self.db.commit()
|
|
except Exception:
|
|
self.db.rollback()
|
|
raise
|
|
|
|
def get_order_info(self, order_id):
|
|
sql = 'SELECT quant, from_address, to_address, create_timestamp FROM orders WHERE order_id = %s'
|
|
logger.debug(f'getting order {order_id} info')
|
|
cursor = self.db.execute_query(sql, [order_id])
|
|
result = cursor.fetchone()
|
|
cursor.close()
|
|
logger.debug(f'order {order_id} info: {result}')
|
|
return result
|
|
|
|
def query(self, order, offset, limit):
|
|
select_sql, params = order.select_sql(
|
|
query_columns=['order_id', 'name', 'phone', 'email', 'quant', 'payment_method', 'from_address',
|
|
'create_timestamp', 'update_timestamp', 'status']
|
|
)
|
|
select_sql = f"{select_sql} LIMIT %s OFFSET %s"
|
|
params = params + [limit, offset]
|
|
cursor = self.db.execute_query(select_sql, params)
|
|
result = cursor.fetchmany(limit)
|
|
cursor.close()
|
|
return result
|