entrouvert
/
passerelle
15
0
Fork 0
Code Issues Pull Requests 15 Releases Activity
passerelle/passerelle/contrib/toulouse_axel/schemas.py

433 lines
15 KiB
Python
Raw Blame History

# passerelle - uniform access to multiple data sources and services
# Copyright (C) 2020 Entr'ouvert
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import copy
import datetime
import os
import re
import xml.etree.ElementTree as ET
from collections import namedtuple
from django.utils.encoding import force_text
import xmlschema
from passerelle.utils.xml import JSONSchemaFromXMLSchema
from . import utils
from .exceptions import AxelError
BASE_XSD_PATH = os.path.join(os.path.dirname(__file__), 'xsd')
class AxelSchema(JSONSchemaFromXMLSchema):
type_map = {
'{urn:AllAxelTypes}DATEREQUIREDType': 'date',
'{urn:AllAxelTypes}DATEType': 'date_optional',
'{urn:AllAxelTypes}OUINONREQUIREDType': 'bool',
'{urn:AllAxelTypes}OUINONType': 'bool_optional',
}
@classmethod
def schema_date(cls):
return {
'type': 'string',
'pattern': '[0-9]{4}-[0-9]{2}-[0-9]{2}',
}
def encode_date(self, obj):
try:
return datetime.datetime.strptime(obj, utils.json_date_format).strftime(utils.xml_date_format)
except ValueError:
return obj
def encode_date_optional(self, obj):
if not obj:
return obj
return self.encode_date(obj)
def decode_date(self, data):
value = datetime.datetime.strptime(data.text, utils.xml_date_format).strftime(utils.json_date_format)
return xmlschema.ElementData(tag=data.tag, text=value, content=data.content, attributes=data.attributes)
def decode_date_optional(self, data):
if not data.text:
return data
return self.decode_date(data)
@classmethod
def schema_bool(cls):
return copy.deepcopy(utils.boolean_type)
def encode_bool(self, obj):
return utils.encode_bool(obj)
def decode_bool(self, data):
value = False
if data.text.lower() == 'oui':
value = True
return xmlschema.ElementData(tag=data.tag, text=value, content=data.content, attributes=data.attributes)
@classmethod
def schema_bool_optional(cls):
schema_bool_optional = cls.schema_bool()
schema_bool_optional['oneOf'].append({'type': 'string', 'enum': ['']})
return schema_bool_optional
def encode_bool_optional(self, obj):
return self.encode_bool(obj)
def decode_bool_optional(self, data):
if not data.text:
return data
return self.decode_bool(data)
def xml_schema_converter(name, root_element):
xsd_path = os.path.join(BASE_XSD_PATH, name)
if not os.path.exists(xsd_path):
return None
return AxelSchema(xsd_path, root_element)
OperationResult = namedtuple('OperationResult', ['json_response', 'xml_request', 'xml_response'])
class Operation(object):
def __init__(self, operation, prefix='Dui/', request_root_element='PORTAIL'):
self.operation = operation
self.request_converter = xml_schema_converter('%sQ_%s.xsd' % (prefix, operation), request_root_element)
self.response_converter = xml_schema_converter('%sR_%s.xsd' % (prefix, operation), 'PORTAILSERVICE')
self.name = re.sub(
'(.?)([A-Z])',
lambda s: s.group(1) + ('-' if s.group(1) else '') + s.group(2).lower(),
operation)
self.snake_name = self.name.replace('-', '_')
@property
def request_schema(self):
schema = self.request_converter.json_schema
schema['flatten'] = True
schema['merge_extra'] = True
return schema
def __call__(self, resource, request_data=None):
client = resource.soap_client()
serialized_request = ''
if self.request_converter:
try:
serialized_request = self.request_converter.encode(request_data)
except xmlschema.XMLSchemaValidationError as e:
raise AxelError('invalid request %s' % str(e))
utils.indent(serialized_request)
serialized_request = force_text(ET.tostring(serialized_request))
try:
self.request_converter.xml_schema.validate(serialized_request)
except xmlschema.XMLSchemaValidationError as e:
raise AxelError(
'invalid request %s' % str(e),
xml_request=serialized_request)
result = client.service.getData(
self.operation,
serialized_request,
'') # FIXME: What is the user parameter for ?
xml_result = ET.fromstring(result.encode('utf-8'))
utils.indent(xml_result)
pretty_result = force_text(ET.tostring(xml_result))
if xml_result.find('RESULTAT/STATUS').text != 'OK':
msg = xml_result.find('RESULTAT/COMMENTAIRES').text
raise AxelError(
msg,
xml_request=serialized_request,
xml_response=pretty_result)
try:
return OperationResult(
json_response=self.response_converter.decode(xml_result),
xml_request=serialized_request,
xml_response=pretty_result
)
except xmlschema.XMLSchemaValidationError as e:
raise AxelError(
'invalid response %s' % str(e),
xml_request=serialized_request,
xml_response=pretty_result)
ref_date_gestion_dui = Operation('RefDateGestionDui')
ref_verif_dui = Operation('RefVerifDui')
ref_famille_dui = Operation('RefFamilleDui')
form_maj_famille_dui = Operation('FormMajFamilleDui')
form_paiement_dui = Operation('FormPaiementDui')
ref_facture_a_payer = Operation('RefFactureAPayer')
ref_facture_pdf = Operation('RefFacturePDF', prefix='')
list_dui_factures = Operation('ListeDuiFacturesPayeesRecettees', request_root_element='LISTFACTURE')
enfants_activites = Operation('EnfantsActivites', request_root_element='DUI')
reservation_periode = Operation('ReservationPeriode')
reservation_annuelle = Operation('ReservationAnnuelle')
PAYMENT_SCHEMA = {
'type': 'object',
'properties': {
'transaction_date': copy.deepcopy(utils.datetime_type),
'transaction_id': {
'type': 'string',
}
},
'required': ['transaction_date', 'transaction_id']
}
LINK_SCHEMA = copy.deepcopy(ref_verif_dui.request_schema['properties']['PORTAIL']['properties']['DUI'])
LINK_SCHEMA['properties'].pop('IDPERSONNE')
LINK_SCHEMA['required'].remove('IDPERSONNE')
UPDATE_FAMILY_FLAGS = {
'maj:adresse': 'ADRESSE',
'maj:rl1': 'RL1',
'maj:rl1_adresse_employeur': 'RL1/ADREMPLOYEUR',
'maj:rl2': 'RL2',
'maj:rl2_adresse_employeur': 'RL2/ADREMPLOYEUR',
'maj:revenus': 'REVENUS',
}
UPDATE_FAMILY_REQUIRED_FLAGS = [
'maj:adresse',
'maj:rl1',
'maj:rl2',
'maj:revenus',
]
for i in range(0, 6):
UPDATE_FAMILY_FLAGS.update({
'maj:enfant_%s' % i: 'ENFANT/%s' % i,
'maj:enfant_%s_sanitaire' % i: 'ENFANT/%s/SANITAIRE' % i,
'maj:enfant_%s_sanitaire_medecin' % i: 'ENFANT/%s/SANITAIRE/MEDECIN' % i,
'maj:enfant_%s_sanitaire_vaccin' % i: 'ENFANT/%s/SANITAIRE/VACCIN' % i,
'maj:enfant_%s_sanitaire_allergie' % i: 'ENFANT/%s/SANITAIRE/ALLERGIE' % i,
'maj:enfant_%s_sanitaire_handicap' % i: 'ENFANT/%s/SANITAIRE/HANDICAP' % i,
'maj:enfant_%s_assurance' % i: 'ENFANT/%s/ASSURANCE' % i,
'maj:enfant_%s_contact' % i: 'ENFANT/%s/CONTACT' % i,
})
UPDATE_FAMILY_REQUIRED_FLAGS.append('maj:enfant_%s' % i)
UPDATE_FAMILY_SCHEMA = copy.deepcopy(
form_maj_famille_dui.request_schema['properties']['PORTAIL']['properties']['DUI'])
for flag in sorted(UPDATE_FAMILY_FLAGS.keys()):
flag_type = copy.deepcopy(utils.boolean_type)
if flag not in UPDATE_FAMILY_REQUIRED_FLAGS:
flag_type['oneOf'].append({'type': 'null'})
flag_type['oneOf'].append({'type': 'string', 'enum': ['']})
UPDATE_FAMILY_SCHEMA['properties'][flag] = flag_type
UPDATE_FAMILY_SCHEMA['required'].append(flag)
UPDATE_FAMILY_SCHEMA['properties'].pop('IDDUI')
UPDATE_FAMILY_SCHEMA['properties'].pop('DATEDEMANDE')
UPDATE_FAMILY_SCHEMA['properties'].pop('QUIACTUALISEDUI')
UPDATE_FAMILY_SCHEMA['required'].remove('IDDUI')
UPDATE_FAMILY_SCHEMA['required'].remove('DATEDEMANDE')
UPDATE_FAMILY_SCHEMA['required'].remove('QUIACTUALISEDUI')
for key in ['IDPERSONNE', 'NOM', 'PRENOM', 'NOMJEUNEFILLE', 'DATENAISSANCE', 'CIVILITE', 'INDICATEURRL']:
UPDATE_FAMILY_SCHEMA['properties']['RL1']['properties'].pop(key)
UPDATE_FAMILY_SCHEMA['properties']['RL1']['required'].remove(key)
UPDATE_FAMILY_SCHEMA['properties']['RL2']['properties'].pop(key)
UPDATE_FAMILY_SCHEMA['properties']['RL2']['required'].remove(key)
UPDATE_FAMILY_SCHEMA['properties']['REVENUS']['properties'].pop('NBENFANTSACHARGE')
UPDATE_FAMILY_SCHEMA['properties']['REVENUS']['required'].remove('NBENFANTSACHARGE')
handicap_fields = [
'AUTREDIFFICULTE',
'ECOLESPECIALISEE',
'INDICATEURAUXILIAIREVS',
'INDICATEURECOLE',
'INDICATEURHANDICAP',
'INDICATEURNOTIFMDPH',
]
sanitaire_properties = UPDATE_FAMILY_SCHEMA['properties']['ENFANT']['items']['properties']['SANITAIRE']['properties']
sanitaire_required = UPDATE_FAMILY_SCHEMA['properties']['ENFANT']['items']['properties']['SANITAIRE']['required']
sanitaire_properties['HANDICAP'] = {
'type': 'object',
'properties': {},
'required': handicap_fields,
}
sanitaire_required.append('HANDICAP')
for key in handicap_fields:
field = sanitaire_properties.pop(key)
sanitaire_properties['HANDICAP']['properties'][key] = field
sanitaire_required.remove(key)
sanitaire_properties.pop('ALLERGIE')
sanitaire_properties['ALLERGIE'] = {
'type': 'object',
'properties': {},
'required': ['ASTHME', 'MEDICAMENTEUSES', 'ALIMENTAIRES', 'AUTRES'],
}
sanitaire_required.append('ALLERGIE')
for key in ['ASTHME', 'MEDICAMENTEUSES', 'ALIMENTAIRES']:
flag_type = copy.deepcopy(utils.boolean_type)
flag_type['oneOf'].append({'type': 'null'})
flag_type['oneOf'].append({'type': 'string', 'enum': ['']})
sanitaire_properties['ALLERGIE']['properties'][key] = flag_type
sanitaire_properties['ALLERGIE']['properties']['AUTRES'] = {
'oneOf': [
{'type': 'null'},
{
'type': 'string',
'minLength': 0,
'maxLength': 50,
}
]
}
UPDATE_FAMILY_SCHEMA['unflatten'] = True
BOOKING_SCHEMA = {
'type': 'object',
'properties': {
'booking_start_date': copy.deepcopy(utils.date_type),
'booking_end_date': copy.deepcopy(utils.date_type),
'booking_list_MAT': {
'oneOf': [
{'type': 'null'},
{
'type': 'array',
'items': {
'type': 'string',
'pattern': '[A-Za-z0-9]+:MAT:[A-Za-z0-9]+:[0-9]{4}-[0-9]{2}-[0-9]{2}',
}
}
]
},
'booking_list_MIDI': {
'oneOf': [
{'type': 'null'},
{
'type': 'array',
'items': {
'type': 'string',
'pattern': '[A-Za-z0-9]+:MIDI:[A-Za-z0-9]+:[0-9]{4}-[0-9]{2}-[0-9]{2}',
}
}
]
},
'booking_list_SOIR': {
'oneOf': [
{'type': 'null'},
{
'type': 'array',
'items': {
'type': 'string',
'pattern': '[A-Za-z0-9]+:SOIR:[A-Za-z0-9]+:[0-9]{4}-[0-9]{2}-[0-9]{2}',
}
}
]
},
'booking_list_GARD': {
'oneOf': [
{'type': 'null'},
{
'type': 'array',
'items': {
'type': 'string',
'pattern': '[A-Za-z0-9]+:GARD:[A-Za-z0-9]+:[0-9]{4}-[0-9]{2}-[0-9]{2}',
}
}
]
},
'child_id': {
'type': 'string',
'minLength': 1,
'maxLength': 8,
},
'regime': {
'oneOf': [
{'type': 'null'},
{'type': 'string', 'enum': ['', 'SV', 'AV']}
]
}
},
'required': ['booking_start_date', 'booking_end_date', 'booking_list_MAT', 'booking_list_MIDI', 'booking_list_SOIR', 'booking_list_GARD', 'child_id']
}
ANNUAL_BOOKING_SCHEMA = {
'type': 'object',
'properties': {
'booking_list_MAT': {
'oneOf': [
{'type': 'null'},
{
'type': 'array',
'items': {
'type': 'string',
'pattern': '[A-Za-z0-9]+:MAT:[A-Za-z0-9]+:(monday|tuesday|wednesday|thursday|friday)',
}
}
]
},
'booking_list_MIDI': {
'oneOf': [
{'type': 'null'},
{
'type': 'array',
'items': {
'type': 'string',
'pattern': '[A-Za-z0-9]+:MIDI:[A-Za-z0-9]+:(monday|tuesday|wednesday|thursday|friday)',
}
}
]
},
'booking_list_SOIR': {
'oneOf': [
{'type': 'null'},
{
'type': 'array',
'items': {
'type': 'string',
'pattern': '[A-Za-z0-9]+:SOIR:[A-Za-z0-9]+:(monday|tuesday|wednesday|thursday|friday)',
}
}
]
},
'booking_list_GARD': {
'oneOf': [
{'type': 'null'},
{
'type': 'array',
'items': {
'type': 'string',
'pattern': '[A-Za-z0-9]+:GARD:[A-Za-z0-9]+:(monday|tuesday|wednesday|thursday|friday)',
}
}
]
},
'child_id': {
'type': 'string',
'minLength': 1,
'maxLength': 8,
},
'regime': {
'oneOf': [
{'type': 'null'},
{'type': 'string', 'enum': ['', 'SV', 'AV']}
]
},
'booking_date': copy.deepcopy(utils.date_type)
},
'required': ['booking_list_MAT', 'booking_list_MIDI', 'booking_list_SOIR', 'booking_list_GARD', 'child_id', 'booking_date']
}
View Git Blame Copy Permalink