refactor(app.py): optimize structure

app.py

@@ -28,6 +28,30 @@ def decorated_function(*args, **kwargs):
         return f(*args, **kwargs)
     return decorated_function
 
+# Common printer communication mixin
+class PrinterCommunicationMixin:
+    def send_zpl_to_printer(self, printer_ip, printer_port, zpl_data):
+        try:
+            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+                sock.settimeout(10)  # Add timeout for better error handling
+                sock.connect((printer_ip, printer_port))
+                sock.sendall(zpl_data.encode('utf-8'))
+        except socket.timeout:
+            logging.error(f"Connection timeout to printer at {printer_ip}:{printer_port}")
+            raise Exception("Printer connection timeout")
+        except socket.error as e:
+            logging.error(f"Socket error while connecting to printer at {printer_ip}:{printer_port}: {str(e)}")
+            raise Exception(f"Printer connection error: {str(e)}")
+        except Exception as e:
+            logging.error(f"Unexpected error while sending data to printer: {str(e)}")
+            raise
+
+    def get_printer_info(self, printer_id):
+        printer = printers.get(printer_id)
+        if not printer:
+            raise ValueError('Printer ID not found')
+        return printer
+
 class PrinterList(Resource):
     method_decorators = [require_apikey]
 
@@ -41,154 +65,134 @@ class PrinterStatus(Resource):
     def get(self):
         status = {}
         for printer_id, printer_info in printers.items():
-            online = self.check_printer_status(printer_info['ip'], printer_info['port'])
-            status[printer_id] = 'Online' if online else 'Offline'
+            try:
+                online = self.check_printer_status(printer_info['ip'], printer_info['port'])
+                status[printer_id] = 'Online' if online else 'Offline'
+            except Exception as e:
+                logging.error(f"Error checking printer {printer_id} status: {str(e)}")
+                status[printer_id] = 'Error'
         return status
 
     def check_printer_status(self, printer_ip, printer_port):
         with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
-            sock.settimeout(5)  # Set timeout to 5 seconds
+            sock.settimeout(5)
             try:
                 sock.connect((printer_ip, printer_port))
                 return True
-            except socket.error as e:
-                print(f"Failed to connect to {printer_ip}:{printer_port} - {e}")
+            except (socket.timeout, socket.error) as e:
+                logging.warning(f"Failed to connect to {printer_ip}:{printer_port} - {e}")
                 return False
 
-class PrintLabel(Resource):
+class PrintLabel(Resource, PrinterCommunicationMixin):
     method_decorators = [require_apikey]
 
     def post(self):
-        errors = validate_request(request.json)
-        if errors:
-            return {'errors': errors}, 400
-
-        data = request.json
-        printer_id = data['printer_id']
-        printer = printers.get(printer_id)
-        if not printer:
-            return {'error': 'Printer ID not found'}, 404
-
-        zpl_command = generate_zpl(**data)
         try:
+            errors = validate_request(request.json)
+            if errors:
+                return {'errors': errors}, 400
+
+            data = request.json
+            printer = self.get_printer_info(data['printer_id'])
+
+            zpl_command = generate_zpl(**data)
             self.send_zpl_to_printer(printer['ip'], printer['port'], zpl_command)
+
             return {'message': 'Label sent to printer successfully'}
+        except ValueError as e:
+            return {'error': str(e)}, 404
         except Exception as e:
+            logging.error(f"Error in PrintLabel: {str(e)}")
             return {'error': str(e)}, 500
 
-    def send_zpl_to_printer(self, printer_ip, printer_port, zpl_data):
-        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
-            sock.connect((printer_ip, printer_port))
-            sock.sendall(zpl_data.encode('utf-8'))
-
-class PrintMsl(Resource):
+class PrintMsl(Resource, PrinterCommunicationMixin):
     method_decorators = [require_apikey]
 
     def post(self):
-        errors = validate_msl_request(request.json)
-        if errors:
-            return {'errors': errors}, 400
-
-        data = request.json
-        printer_id = data['printer_id']
-        printer = printers.get(printer_id)
-        if not printer:
-            return {'error': 'Printer ID not found'}, 404
-
-        print_msl_command = generate_msl_sticker(**data)
         try:
+            errors = validate_msl_request(request.json)
+            if errors:
+                return {'errors': errors}, 400
+
+            data = request.json
+            printer = self.get_printer_info(data['printer_id'])
+
+            print_msl_command = generate_msl_sticker(**data)
             self.send_zpl_to_printer(printer['ip'], printer['port'], print_msl_command)
+
             return {'message': 'MSL label sent to printer successfully'}
+        except ValueError as e:
+            return {'error': str(e)}, 404
         except Exception as e:
+            logging.error(f"Error in PrintMsl: {str(e)}")
             return {'error': str(e)}, 500
 
-    def send_zpl_to_printer(self, printer_ip, printer_port, zpl_data):
-        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
-            sock.connect((printer_ip, printer_port))
-            sock.sendall(zpl_data.encode('utf-8'))
-
-
-class PrintSpecialInstructions(Resource):
+class PrintSpecialInstructions(Resource, PrinterCommunicationMixin):
     method_decorators = [require_apikey]
 
     def post(self):
-        errors = validate_special_instructions_request(request.json)
-        if errors:
-            return {'errors': errors}, 400
-
-        data = request.json
-        printer_id = data['printer_id']
-        printer = printers.get(printer_id)
-        if not printer:
-            return {'error': 'Printer ID not found'}, 404
-
-        print_msl_command = generate_special_instructions_label(**data)
         try:
-            self.send_zpl_to_printer(printer['ip'], printer['port'], print_msl_command)
+            errors = validate_special_instructions_request(request.json)
+            if errors:
+                return {'errors': errors}, 400
+
+            data = request.json
+            printer = self.get_printer_info(data['printer_id'])
+
+            print_command = generate_special_instructions_label(**data)
+            self.send_zpl_to_printer(printer['ip'], printer['port'], print_command)
+
             return {'message': 'Special Instructions label sent to printer successfully'}
+        except ValueError as e:
+            return {'error': str(e)}, 404
         except Exception as e:
+            logging.error(f"Error in PrintSpecialInstructions: {str(e)}")
             return {'error': str(e)}, 500
 
-    def send_zpl_to_printer(self, printer_ip, printer_port, zpl_data):
-        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
-            sock.connect((printer_ip, printer_port))
-            sock.sendall(zpl_data.encode('utf-8'))
-
-
-class PrintDry(Resource):
+class PrintDry(Resource, PrinterCommunicationMixin):
     method_decorators = [require_apikey]
 
     def post(self):
-        errors = validate_dry_request(request.json)
-        if errors:
-            return {'errors': errors}, 400
-
-        data = request.json
-        printer_id = data['printer_id']
-        printer = printers.get(printer_id)
-        if not printer:
-            return {'error': 'Printer ID not found'}, 404
-
-        print_msl_command = generate_dry_label(**data)
         try:
-            self.send_zpl_to_printer(printer['ip'], printer['port'], print_msl_command)
+            errors = validate_dry_request(request.json)
+            if errors:
+                return {'errors': errors}, 400
+
+            data = request.json
+            printer = self.get_printer_info(data['printer_id'])
+
+            print_command = generate_dry_label(**data)
+            self.send_zpl_to_printer(printer['ip'], printer['port'], print_command)
+
             return {'message': 'DRY label sent to printer successfully'}
+        except ValueError as e:
+            return {'error': str(e)}, 404
         except Exception as e:
+            logging.error(f"Error in PrintDry: {str(e)}")
             return {'error': str(e)}, 500
 
-    def send_zpl_to_printer(self, printer_ip, printer_port, zpl_data):
-        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
-            sock.connect((printer_ip, printer_port))
-            sock.sendall(zpl_data.encode('utf-8'))
-
-
-class PrintTracescanLabel(Resource):
+class PrintTracescanLabel(Resource, PrinterCommunicationMixin):
     method_decorators = [require_apikey]
 
     def post(self):
-        errors = validate_tracescan_request(request.json)
-        if errors:
-            return {'errors': errors}, 400
-
-        data = request.json
-        printer_id = data['printer_id']
-        printer = printers.get(printer_id)
-        if not printer:
-            return {'error': 'Printer ID not found'}, 404
-
-        print_msl_command = generate_tracescan_label(**data)
         try:
-            self.send_zpl_to_printer(printer['ip'], printer['port'], print_msl_command)
+            errors = validate_tracescan_request(request.json)
+            if errors:
+                return {'errors': errors}, 400
+
+            data = request.json
+            printer = self.get_printer_info(data['printer_id'])
+
+            print_command = generate_tracescan_label(**data)
+            self.send_zpl_to_printer(printer['ip'], printer['port'], print_command)
+
             return {'message': 'Tracescan label sent to printer successfully'}
+        except ValueError as e:
+            return {'error': str(e)}, 404
         except Exception as e:
+            logging.error(f"Error in PrintTracescanLabel: {str(e)}")
             return {'error': str(e)}, 500
 
-    def send_zpl_to_printer(self, printer_ip, printer_port, zpl_data):
-        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
-            sock.connect((printer_ip, printer_port))
-            sock.sendall(zpl_data.encode('utf-8'))
-
-
 class HelloWorld(Resource):
     def get(self):
         return {'message': 'miniprint api'}