实现多个运输API的物流项目

Question

我有一个Python逻辑项目，它使用类继承实现了几个交付API。这些类中的每一个都必须做三件事：

• 使用适当的参数向每个端点发出请求
• 序列化和映射原始输入以存储它
• 最后，根据每一家公司的规格，制作一份文件。

首先，我认为把所有这些东西放在一个令人印象深刻的文件中是不可能的，尽管它处理的任务非常不相关，比如发送HTTP请求和格式化PDF文件。最后，我看到了这个目录结构：

shipment/
  shipment.py - primitive Shipment class (for mapping data)
  printer.py - primitive Printer class (for producing documents)
  controller.py - primitives for sending requests, inherits Shipment and has Printer instance as member

providers/
  provider1/
    shipment.py - implements shipment.Shipment
    printer.py - implements shipment.Printer
    controller.py - implements shipment.Controller
  provider2/
    shipment.py
    printer.py
    controller.py

如您所见，每个送货提供程序类都被拆分为3个单独的文件。Printer由Controller继承，并提供"summary_to_pdf“方法。

装运/控制.：

import logging
from abc import ABCMeta, abstractmethod
from datetime import datetime
from os import environ
from bson.objectid import ObjectId
from lib.database import db, ReturnDocument
from lib.model import Shipment, IndividualShipment, Sale

def ProviderClass(mapping_name:str, shipment_class:type, printer_class:type, url:str, packets_limit:int=0, asynchronous:bool=False):
    def decorator(cls):
        cls.mapping_name = mapping_name
        cls.Shipment = shipment_class
        cls.Printer = printer_class
        cls.url = url
        cls.packets_limit = packets_limit
        cls.asynchronous = asynchronous
        class NewClass(cls, printer_class, ShipmentController):
            __name__ = cls.__name__
            pass

        return NewClass
    return decorator

def ProviderInit():
    def decorator(init):
        def new_init(self, **kwargs):
            self.profile = kwargs
            self.Printer.__init__(self, **kwargs)
            ShipmentController.__init__(self, **kwargs)
            init(self, **kwargs)

        return new_init
    return decorator

class ShipmentHealthException(metaclass=ABCMeta):
    def __init__(self, message:str):
        self.message = message
    def __str__(self):
        return f'Shipment health check failed: {message}'

class ShipmentController:
    def __init__(self, **kwargs):
        self.raw_posting_objects = []
        self.logger = logging.getLogger('shipment')

    def build_shipment(self, **kwargs):
        self.raw_posting_objects = kwargs['posting_objects']
        return self.Shipment(**{ **kwargs, **self.profile, 'sender': self.profile['business'] })

    def insert(self, shipment, stamp:str, local_id:str, foreign_id:str):
        payload = {
            'shipment_profile': self.profile['_id'],
            'created_at': datetime.now(),
            'stamp': stamp,
            'local_id': local_id,
            'foreign_id': foreign_id,
            'objects_count': len(shipment)
        }

        shipment_id = Shipment.insert(**payload).inserted_id
        for s in shipment:
            db['sales'].find_one_and_update({ '_id': s['_id'] }, { '$set': { 'shipment': shipment_id } })

        return shipment_id

    @classmethod
    def _get_id(cls):
        return db['shipments'].count_documents({})

    @classmethod
    def _get_individual_id(cls):
        _id = 100000 + db['sales'].count_documents({
            'local_individual_shipment_id': {
                '$ne': None
            }
        })
        return str(_id)

    @staticmethod
    def instantiate(provider_mapping, **kwargs):
        return provider_mapping[kwargs['provider']['controller']](**kwargs)

    @classmethod
    def _raise_not_implemented(cls):
        raise NotImplementedError('Method not implemented for this provider ({})'.format(cls.__name__))

    @abstractmethod
    def check_health(self):
        '''Checks whether provider is operational or not
        Implemented in:
            - Correios
        '''
        self._raise_not_implemented()

    @abstractmethod
    def get_id(self):
        '''Client unique ID for shipment provider
        Implemented in:
            - Correios
            - Jadlog
        '''
        self._raise_not_implemented()

    @abstractmethod
    def get_labels(self, **kwargs):
        '''Requests provider for label (tracking code) numbers
        Implemented in:
            - Correios
        '''
        self._raise_not_implemented()

    @abstractmethod
    def append_verifier_digit(self, **kwargs):
        '''Appends verifier digits in label numbers
        Implemented in:
            - Correios
        '''
        self._raise_not_implemented()

    @abstractmethod
    def get_services(self, save_to_db=False, **kwargs):
        '''Asks provider for available shipment services
        Implemented in:
            - Correios
            - Jadlog
        '''
        self._raise_not_implemented()

    def request_already_sent_shipment(self, _list, local_shipment_id, service_id):
        unsent = [ sale for sale in _list if not sale.get('individual_shipment') ]
        for sale in unsent:
            self.request_individual_shipment(sale, sale['_id'])

        return local_shipment_id, self.build_shipment(posting_objects=_list)

    @abstractmethod
    def request_shipment(self, **kwargs):
        '''Sends PLP to be acquited by provider
        Implemented in:
            - Correios
        '''
        return self.request_already_sent_shipment(**kwargs)

    @abstractmethod
    def _request_individual_shipment(self, **kwargs):
        '''Requests a single object shipment (in case of asynchronous shipment)
        Implemented in:
            - Jadlog
        '''
        self._raise_not_implemented()

    def request_individual_shipment(self, item:dict, item_id, **kwargs):
        query = {
            '_id': ObjectId(item_id),
            'individual_shipment': None
        }

        sale = Sale.find_one(**query)
        if not sale:
            raise Exception('order already dispatched')

        foreign_id, code = self._request_individual_shipment(item, **kwargs)
        foreign_id = str(foreign_id)

        local_id = self._get_individual_id()

        individual_shipment = IndividualShipment.insert(local_id=local_id, foreign_id=foreign_id, code=code).inserted_id

        item = db['sales'].find_one_and_update({ '_id': sale['_id'] }, { '$set': { 'individual_shipment': individual_shipment } }, return_document=ReturnDocument.AFTER)
        return foreign_id, local_id, code

    @abstractmethod
    def request_updated_shipment(self, **kwargs):
        '''Retrieves previous sent PLP
        Implemented in:
            - Correios
        '''
        self._raise_not_implemented()

请随时发表意见，全面的评论以及。

Answer 1

在我看来，您的代码是合理的，您的实现基于特定用途的通用对象。

我可能会做的不同的是整体架构。是否需要将该功能作为一个单独的过程实现？考虑到不同的需求，我想我应该把它构建成一个由微服务组成的系统，通过队列系统(例如RabbitMQ)进行协作。然后，每个程序变成一个简单的读输入循环，处理输入，根据处理结果将结果发送到适当的下一个处理器。选择下一个处理器的最后一个子步骤可以在交换机处理器中完成，它还可以提供系统工作情况的视图。

微服务体系结构将具有简化的代码，但您的部署将变得更加复杂，因为您必须成功地部署所有的微服务处理器才能“启动”系统。要关闭系统，首先需要排完所有队列。但是，如果要确保在关闭之前处理所有读取的数据，则控制关闭将需要统一程序中类似的内容。