365 lines
13 KiB
Python
365 lines
13 KiB
Python
# 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 <http://www.gnu.org/licenses/>.
|
|
|
|
'''Gateway to API-Particulier web-service from SGMAP:
|
|
https://particulier.api.gouv.fr/
|
|
'''
|
|
|
|
import re
|
|
from collections import OrderedDict
|
|
from urllib import parse
|
|
|
|
import requests
|
|
|
|
try:
|
|
from json.decoder import JSONDecodeError
|
|
except ImportError:
|
|
JSONDecodeError = ValueError
|
|
|
|
from django.contrib.postgres.fields import ArrayField
|
|
from django.core.cache import cache
|
|
from django.db import models
|
|
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 passerelle.utils.validation import is_number
|
|
|
|
from .known_errors import KNOWN_ERRORS
|
|
|
|
# regular expression for numero_fiscal and reference_avis
|
|
FISCAL_RE = re.compile(r'^[0-9a-zA-Z]{13}$')
|
|
|
|
|
|
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
|
|
|
|
accessible_scopes = ArrayField(
|
|
models.CharField(max_length=32),
|
|
null=True,
|
|
blank=True,
|
|
editable=False,
|
|
)
|
|
|
|
hide_description_fields = ['accessible_scopes']
|
|
|
|
@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(
|
|
'API-particulier platform "%s" connection error: %s' % (self.platform, exception_to_text(e)),
|
|
log_error=True,
|
|
data={
|
|
'code': 'connection-error',
|
|
'platform': self.platform,
|
|
'error': str(e),
|
|
},
|
|
)
|
|
try:
|
|
data = response.json()
|
|
except JSONDecodeError as e:
|
|
content = repr(response.content[:1000])
|
|
raise APIError(
|
|
'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': str(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, []):
|
|
error_data = {
|
|
'code': data.get('error', 'api-error').replace('_', '-'),
|
|
'status_code': response.status_code,
|
|
'platform': self.platform,
|
|
'content': data,
|
|
}
|
|
if response.status_code // 100 == 4:
|
|
# for 40x errors, allow caching of the response, as it should not change with future requests
|
|
return {
|
|
'err': 1,
|
|
'err_desc': message,
|
|
'data': error_data,
|
|
}
|
|
raise APIError(message, data=error_data)
|
|
raise APIError(
|
|
'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,
|
|
}
|
|
|
|
def get_scopes(self):
|
|
response = self.get('introspect')
|
|
result = sorted(response['data'].get('scopes'))
|
|
return result
|
|
|
|
def clean(self):
|
|
try:
|
|
scopes = self.get_scopes()
|
|
except APIError:
|
|
scopes = []
|
|
self.accessible_scopes = [x[:32] for x in scopes]
|
|
|
|
def daily(self):
|
|
super().daily()
|
|
try:
|
|
scopes = self.get_scopes()
|
|
except APIError:
|
|
scopes = []
|
|
self.accessible_scopes = [x[:32] for x in scopes]
|
|
self.save()
|
|
|
|
@endpoint(
|
|
perm='can_access',
|
|
description=_('Get scopes available'),
|
|
display_order=1,
|
|
)
|
|
def scopes(self, request):
|
|
scopes = self.get_scopes()
|
|
self.accessible_scopes = [x[:32] for x in scopes]
|
|
self.save()
|
|
return {
|
|
'err': 0,
|
|
'data': scopes,
|
|
}
|
|
|
|
@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):
|
|
# clean the references
|
|
numero_fiscal = re.sub(r'\s', '', numero_fiscal)[:13]
|
|
reference_avis = re.sub(r'\s', '', reference_avis)[:13]
|
|
# check they are both 13 digits numbers
|
|
if not FISCAL_RE.match(numero_fiscal):
|
|
raise APIError(
|
|
'bad numero_fiscal, it MUST be composed of 13 alphanumeric characters', status_code=400
|
|
)
|
|
if not FISCAL_RE.match(reference_avis):
|
|
raise APIError(
|
|
'bad reference_avis, it MUST be composed of 13 alphanumeric characters', status_code=400
|
|
)
|
|
|
|
cache_key = 'svai-%s-%s-%s' % (self.pk, numero_fiscal, reference_avis)
|
|
data = cache.get(cache_key)
|
|
if data is None:
|
|
data = self.get(
|
|
'v2/avis-imposition',
|
|
params={
|
|
'numeroFiscal': numero_fiscal,
|
|
'referenceAvis': reference_avis,
|
|
},
|
|
user=user,
|
|
)
|
|
# put in cache for two hours
|
|
cache.set(cache_key, data, 3600 * 2)
|
|
else:
|
|
self.logger.info('found response in cache, "%s"', data)
|
|
return data
|
|
|
|
@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')
|
|
if len(numero_allocataire) != 7 or not is_number(numero_allocataire):
|
|
raise APIError('numero_allocataire should be a 7 digits number')
|
|
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')
|