devicehub-django/dpp/api_dlt.py

152 lines
3.4 KiB
Python
Raw Normal View History

import time
import logging
from django.conf import settings
2024-11-15 17:56:11 +00:00
from ereuseapi.methods import API
from dpp.models import Proof, Dpp
logger = logging.getLogger('django')
# """The code of the status response of api dlt."""
STATUS_CODE = {
"Success": 201,
"Notwork": 400
}
ALGORITHM = "sha3-256"
PROOF_TYPE = {
'Register': 'Register',
'IssueDPP': 'IssueDPP',
'proof_of_recycling': 'proof_of_recycling',
'Erase': 'Erase',
'EWaste': 'EWaste',
}
2024-11-15 17:56:11 +00:00
def connect_api():
if not settings.get('TOKEN_DLT'):
2024-11-15 17:56:11 +00:00
return
token_dlt = settings.get('TOKEN_DLT')
api_dlt = settings.get('API_DLT')
2024-11-15 17:56:11 +00:00
return API(api_dlt, token_dlt, "ethereum")
def register_dlt(chid, phid, proof_type=None):
api = connect_api()
2024-11-15 17:56:11 +00:00
if not api:
return
if proof_type:
return api.generate_proof(
chid,
2024-11-15 17:56:11 +00:00
ALGORITHM,
phid,
proof_type,
settings.get('ID_FEDERATED')
2024-11-15 17:56:11 +00:00
)
return api.register_device(
chid,
ALGORITHM,
phid,
settings.get('ID_FEDERATED')
)
def issuer_dpp_dlt(dpp):
phid = dpp.split(":")[0]
api = connect_api()
if not api:
return
return api.issue_passport(
dpp,
ALGORITHM,
phid,
settings.get('ID_FEDERATED')
)
2024-11-15 17:56:11 +00:00
def save_proof(signature, ev_uuid, result, proof_type, user):
if result['Status'] == STATUS_CODE.get("Success"):
2024-11-15 17:56:11 +00:00
timestamp = result.get('Data', {}).get('data', {}).get('timestamp')
if not timestamp:
return
d = {
"type": proof_type,
2024-11-15 17:56:11 +00:00
"timestamp": timestamp,
"issuer": user.institution.id,
"user": user,
"uuid": ev_uuid,
"signature": signature,
2024-11-15 17:56:11 +00:00
}
Proof.objects.create(**d)
2024-11-15 17:56:11 +00:00
def register_device_dlt(chid, phid, ev_uuid, user):
cny_a = 1
while cny_a:
result = register_dlt(chid, phid)
try:
assert result['Status'] == STATUS_CODE.get("Success")
assert result['Data']['data']['timestamp']
cny_a = 0
except Exception:
if result.get("Data") != "Device already exists":
logger.error("API return: %s", result)
time.sleep(10)
else:
cny_a = 0
save_proof(phid, ev_uuid, result, PROOF_TYPE['Register'], user)
# TODO is neccesary?
# if settings.get('ID_FEDERATED'):
# cny = 1
# while cny:
# try:
# api.add_service(
# chid,
# 'DeviceHub',
# settings.get('ID_FEDERATED'),
# 'Inventory service',
# 'Inv',
# )
# cny = 0
# except Exception:
# time.sleep(10)
def register_passport_dlt(chid, phid, ev_uuid, user):
dpp = "{chid}:{phid}".format(chid=chid, phid=phid)
if Proof.objects.filter(signature=dpp, type=PROOF_TYPE['IssueDPP']).exists():
2024-11-15 17:56:11 +00:00
return
cny_a = 1
while cny_a:
try:
result = issuer_dpp_dlt(dpp)
cny_a = 0
except Exception as err:
logger.error("ERROR API issue passport return: %s", err)
time.sleep(10)
if result['Status'] is not STATUS_CODE.get("Success"):
logger.error("ERROR API issue passport return: %s", result)
return
save_proof(phid, ev_uuid, result, PROOF_TYPE['IssueDPP'], user)