# passerelle - uniform access to multiple data sources and services # Copyright (C) 2017 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 . '''Gateway to API-Particulier web-service from SGMAP: https://particulier.api.gouv.fr/ ''' from collections import OrderedDict import requests try: from json.decoder import JSONDecodeError except ImportError: JSONDecodeError = ValueError from django.db import models from django.utils import six from django.utils.six.moves.urllib import parse from django.utils.translation import ugettext_lazy as _ from passerelle.base.models import BaseResource from passerelle.utils.api import endpoint from passerelle.utils.conversion import exception_to_text from passerelle.utils.jsonresponse import APIError from .known_errors import KNOWN_ERRORS class APIParticulier(BaseResource): PLATFORMS = [ {'name': 'prod', 'label': _('Production'), 'url': 'https://particulier.api.gouv.fr/api/'}, { 'name': 'test', 'label': _('Test'), 'url': 'https://particulier-test.api.gouv.fr/api/', }, ] PLATFORMS = OrderedDict([(platform['name'], platform) for platform in PLATFORMS]) platform = models.CharField( verbose_name=_('Platform'), max_length=8, choices=[(key, platform['label']) for key, platform in PLATFORMS.items()], ) api_key = models.CharField(max_length=256, default='', blank=True, verbose_name=_('API key')) log_requests_errors = False @property def url(self): return self.PLATFORMS[self.platform]['url'] def get(self, path, **kwargs): user = kwargs.pop('user', None) url = parse.urljoin(self.url, path) headers = {'X-API-KEY': self.api_key} if user: headers['X-User'] = user try: response = self.requests.get(url, headers=headers, timeout=5, **kwargs) except requests.RequestException as e: raise APIError( u'API-particulier platform "%s" connection error: %s' % (self.platform, exception_to_text(e)), log_error=True, data={ 'code': 'connection-error', 'platform': self.platform, 'error': six.text_type(e), }, ) try: data = response.json() except JSONDecodeError as e: content = repr(response.content[:1000]) raise APIError( u'API-particulier platform "%s" returned non-JSON content with status %s: %s' % (self.platform, response.status_code, content), log_error=True, data={ 'code': 'non-json', 'status_code': response.status_code, 'exception': six.text_type(e), 'platform': self.platform, 'content': content, }, ) if response.status_code != 200: # avoid logging http errors about non-transport failure message = data.get('message', '') if message in KNOWN_ERRORS.get(response.status_code, []): raise APIError( message, data={ 'code': data.get('error', 'api-error').replace('_', '-'), 'status_code': response.status_code, 'platform': self.platform, 'content': data, }, ) raise APIError( u'API-particulier platform "%s" returned a non 200 status %s: %s' % (self.platform, response.status_code, data), log_error=True, data={ 'code': 'non-200', 'status_code': response.status_code, 'platform': self.platform, 'content': data, }, ) return { 'err': 0, 'data': data, } @endpoint( perm='can_access', show=False, description=_('Get citizen\'s fiscal informations'), parameters={ 'numero_fiscal': { 'description': _('fiscal identifier'), 'example_value': '1562456789521', }, 'reference_avis': { 'description': _('tax notice number'), 'example_value': '1512456789521', }, 'user': { 'description': _('requesting user'), 'example_value': 'John Doe (agent)', }, }, ) def impots_svair(self, request, numero_fiscal, reference_avis, user=None): # deprecated endpoint return self.v2_avis_imposition(request, numero_fiscal, reference_avis, user=user) @endpoint( name='avis-imposition', perm='can_access', description=_('Get citizen\'s fiscal informations'), parameters={ 'numero_fiscal': { 'description': _('fiscal identifier'), 'example_value': '1562456789521', }, 'reference_avis': { 'description': _('tax notice number'), 'example_value': '1512456789521', }, 'user': { 'description': _('requesting user'), 'example_value': 'John Doe (agent)', }, }, json_schema_response={ 'type': 'object', 'required': ['err'], 'properties': { 'err': {'enum': [0, 1]}, 'declarant1': { 'type': 'object', 'properties': { 'nom': {'type': 'string'}, 'nomNaissance': {'type': 'string'}, 'prenoms': {'type': 'string'}, 'dateNaissance': {'type': 'string'}, }, }, 'declarant2': { 'type': 'object', 'properties': { 'nom': {'type': 'string'}, 'nomNaissance': {'type': 'string'}, 'prenoms': {'type': 'string'}, 'dateNaissance': {'type': 'string'}, }, }, 'foyerFiscal': { 'type': 'object', 'properties': { 'annee': {'type': 'integer'}, 'adresse': {'type': 'string'}, }, }, 'dateRecouvrement': {'type': 'string', 'pattern': r'^\d{1,2}/\d{1,2}/\d{4}$'}, 'dateEtablissement': {'type': 'string', 'pattern': r'^\d{1,2}/\d{1,2}/\d{4}$'}, 'nombreParts': {'type': 'integer'}, 'situationFamille': {'type': 'string'}, 'nombrePersonnesCharge': {'type': 'integer'}, 'revenuBrutGlobal': {'type': 'integer'}, 'revenuImposable': {'type': 'integer'}, 'impotRevenuNetAvantCorrections': {'type': 'integer'}, 'montantImpot': {'type': 'integer'}, 'revenuFiscalReference': {'type': 'integer'}, 'anneeImpots': {'type': 'string', 'pattern': r'^[0-9]{4}$'}, 'anneeRevenus': {'type': 'string', 'pattern': r'^[0-9]{4}$'}, 'erreurCorrectif': {'type': 'string'}, 'situationPartielle': {'type': 'string'}, }, }, ) def v2_avis_imposition(self, request, numero_fiscal, reference_avis, user=None): numero_fiscal = numero_fiscal.strip()[:13] reference_avis = reference_avis.strip()[:13] if len(numero_fiscal) < 13 or len(reference_avis) < 13: raise APIError('bad numero_fiscal or reference_avis, must be 13 chars long', status_code=400) return self.get( 'v2/avis-imposition', params={ 'numeroFiscal': numero_fiscal, 'referenceAvis': reference_avis, }, user=user, ) @endpoint( perm='can_access', show=False, description=_('Get family allowances recipient informations'), parameters={ 'code_postal': { 'description': _('postal code'), 'example_value': '99148', }, 'numero_allocataire': { 'description': _('recipient identifier'), 'example_value': '0000354', }, 'user': { 'description': _('requesting user'), 'example_value': 'John Doe (agent)', }, }, ) def caf_famille(self, request, code_postal, numero_allocataire, user=None): # deprecated endpoint return self.v2_situation_familiale(request, code_postal, numero_allocataire, user=user) @endpoint( name='situation-familiale', perm='can_access', description=_('Get family allowances recipient informations'), parameters={ 'code_postal': { 'description': _('postal code'), 'example_value': '99148', }, 'numero_allocataire': { 'description': _('recipient identifier'), 'example_value': '0000354', }, 'user': { 'description': _('requesting user'), 'example_value': 'John Doe (agent)', }, }, ) def v2_situation_familiale(self, request, code_postal, numero_allocataire, user=None): if not code_postal.strip() or not numero_allocataire.strip(): raise APIError('missing code_postal or numero_allocataire', status_code=400) return self.get( 'v2/composition-familiale', params={ 'codePostal': code_postal, 'numeroAllocataire': numero_allocataire, }, user=user, ) category = _('Business Process Connectors') class Meta: verbose_name = _('API Particulier')