Untitled
unknown
python
a year ago
2.5 kB
6
Indexable
Never
# file: views.py @router.post('/orders/tracking') def order_tracking(msg: messages.UpdateOrderTrackingRequest): with Context.fastapi(): domain.order.update_order_details(msg) return {'Status': True} # ============ # file: messages.py class UpdateOrderTrackingRequest(util.NoonBaseModel): order_code: str order_status_code: str driver_name: str | None driver_phone: str | None driver_longitude: float | None driver_latitude: float | None # ============ # util.py def get_darkstore_by_order_code(order_code): darkstore = sql(engine, ''' SELECT id_darkstore, darkstore_code, load_factor, latitude, longitude FROM darkstore LEFT JOIN `order` ON darkstore.id_darkstore = `order`.id_darkstore WHERE order_code=:order_code ''', order_code=order_code).dict() assert darkstore, f"No darkstore found" return darkstore # ====================== # file: order.py # [{'order_code': 'X00000001', 'order_status_code': 'DELIVERING', 'longitude': None, 'latitude': None, 'name': 'Mr.Driver 1', 'phone_number': '056123'}] def update_order_details(order_tracking): id_order_status = libminutes.models.util.get_order_status_by_code(order_tracking['order_status_code'])[ 'id_order_status'] eta = None if order_tracking['driver_longitude']: source_location = { 'longitude': order_tracking['driver_longitude'], 'latitude': order_tracking['driver_latitude'] } darkstore = libminutes.models.util.get_darkstore_by_order_code(order_tracking['order_code']) darkstore_load_factor = darkstore['load_factor'] target_location = libminutes.models.util.get_customer_order_location_id_address(order_tracking['id_address']) eta = libminutes.models.util.get_eta(source_location, target_location, darkstore_load_factor) row = { 'order_code': order_tracking['order_code'], 'driver_latitude': order_tracking['driver_longitude'], 'driver_longitude': order_tracking['driver_latitude'], 'id_order_status': id_order_status, 'eta': eta } update_columns = ['driver_latitude', 'driver_longitude', 'id_order_status', 'eta'] unique_columns = ['order_code'] sqlutil.upsert( engine, tables.Order, [row], unique_columns=unique_columns, update_columns=update_columns )