SOAP сервис со списком общих баз для 1С из файла

1c_ibases_soap.py

import hashlib
import logging
import os
import uuid
from dataclasses import dataclass, field
from typing import Optional, Dict, Union

from spyne import Application, rpc, ServiceBase, Unicode, Boolean, Integer
from spyne.protocol.soap import Soap11
from spyne.server.wsgi import WsgiApplication

infobases_filename = 'ibases.v8i'
NULL_UID = str(uuid.UUID(int=0))

infobases: Optional['InfobasesState'] = None

DistrDict = Dict[str, Union[str, 'DistrDict']]
distribs: DistrDict = {
    '8.3.18.1289': {
        'windows': {
            'x86': {
                'file': '/path/distr_x86.zip',
                'url': 'http://server/1c/distr_x86.zip'
            },
            'x86_64': {
                'file': '/path/distr_x64.zip',
                'url': 'http://server/1c/distr_x64.zip'
            },
        },
        'linux': {},
        'linuxdeb': {},
        'linuxrpm': {},
        'macos': {},
    }
}

@dataclass(frozen=True)
class InfobasesState:
    text: str
    mtime: float
    check_code: str = field(init=False)

    def __post_init__(self):
        object.__setattr__(self, "check_code", self.calc_hash())

    def calc_hash(self) -> str:
        _hash = hashlib.md5(self.text.encode(encoding='utf-8'))
        return str(uuid.UUID(bytes=_hash.digest()))

    def is_changed(self, chek_code: str) -> bool:
        return chek_code != self.check_code

    @staticmethod
    def get() -> 'InfobasesState':
        global infobases

        def load() -> str:
            with open(infobases_filename, "rt", encoding="utf-8") as f:
                return f.read()

        cur_mtime = os.path.getmtime(infobases_filename)
        if infobases is None or infobases.mtime != cur_mtime:
            logging.info(f'list file changed, mtime: {cur_mtime}')
            infobases = InfobasesState(load(), cur_mtime)
        return infobases


def new_client_id():
    return str(uuid.uuid4())


class WebDistributiveLocation(ServiceBase):
    @rpc(Unicode(sub_name="DistributiveType"),
         Unicode(sub_name="Arch"),
         Unicode(sub_name="Version"),
         _returns=[Unicode(sub_name="return", nillable=True),
                   Integer(sub_name="Size"),
                   Unicode(sub_name="URL"),])
    def GetDistributiveInfo(ctx, os_type:str, arch: str, version: str):
        distr: Optional[DistrDict] = None
        try:
            distr = distribs[version][os_type.lower()][arch]
        except KeyError:
            pass

        if distr is None:
            return "", 0, ""
        
        url: str = ''
        size: int = 0

        try:
            url = distr['url']
            size = os.path.getsize(distr['file'])
        except Exception:
            pass

        return "", size, url


class WebCommonInfoBases(ServiceBase):
    @rpc(Unicode(sub_name="ClientID"),
         _returns=[Unicode(sub_name="return", nillable=True),
                   Unicode(sub_name="ClientID"),
                   Unicode(sub_name="InfoBaseCheckCode", nillable=False, min_occurs=1),
                   Unicode(sub_name="InfoBases", nillable=False, min_occurs=1)])
    def GetInfoBases(ctx, client_id):
        ibases = InfobasesState.get()
        ClientID = new_client_id() if client_id == NULL_UID else client_id
        InfoBases = ibases.text
        InfoBaseCheckCode = ibases.check_code

        return "", ClientID, InfoBaseCheckCode, InfoBases

    @rpc(Unicode(sub_name="ClientID"),
         Unicode(sub_name="InfoBaseCheckCode"),
         _returns=[Unicode(sub_name="return", nillable=True),
                   Boolean(sub_name="InfoBasesChanged", nillable=False, min_occurs=1),
                   Unicode(sub_name="URL", nillable=False, min_occurs=1)])
    def CheckInfoBases(ctx, _, check_code):
        ibases = InfobasesState.get()
        InfoBasesChanged = ibases.is_changed(check_code)
        URL = "http://localhost:8000/WebCommonInfoBases"
        return "", InfoBasesChanged, URL


application = Application(services=[WebCommonInfoBases, WebDistributiveLocation],
                          name='WebCommonInfoBasesSoap',
                          tns='listservice.1c',
                          in_protocol=Soap11(validator='lxml'),
                          out_protocol=Soap11())

wsgi_application = WsgiApplication(application)


def main():
    import logging

    from wsgiref.simple_server import make_server

    logging.basicConfig(level=logging.DEBUG)
    logging.getLogger('spyne.protocol.xml').setLevel(logging.DEBUG)

    logging.info("listening to http://127.0.0.1:8000")
    logging.info("wsdl is at: http://localhost:8000/?wsdl")

    server = make_server('127.0.0.1', 8000, wsgi_application)
    server.serve_forever()


if __name__ == '__main__':
    main()