KumiSystems/pyadonis
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

More functions, blacked

Browse source
This commit is contained in:
Kumi 2023-07-03 11:05:13 +02:00
parent 803878eca1
commit 2c7229d58e
Signed by: kumi
GPG key ID: ECBCC9082395383F
7 changed files with 154 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
pyproject.toml
View file

@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "pyadonis"
version = "0.9.5"
version = "0.9.6"
authors = [
{ name="Kumi Systems e.U.", email="office@kumi.systems" },
]

3
requirements.txt
View file

@ -1 +1,2 @@
pillow
pillow
python-dateutil

165
src/pyadonis/classes/adonis.py
View file

@ -15,7 +15,13 @@ INTEGRATION = 1
class Adonis:
def __init__(self, crew_portal_base_url: str, integration_base_url: str, login: str, password: str):
def __init__(
self,
crew_portal_base_url: str,
integration_base_url: str,
login: str,
password: str,
):
self.crew_portal_base_url = crew_portal_base_url
self.integration_base_url = integration_base_url
self.login = login
@ -23,10 +29,27 @@ class Adonis:
@classmethod
def fromConfig(cls, config):
return cls(config.crew_portal_base_url, config.integration_base_url, config.login, config.password)
return cls(
config.crew_portal_base_url,
config.integration_base_url,
config.login,
config.password,
)
def request(self, endpoint: str, payload: dict, add_token: bool = True, wrap_request: bool = True, extract_response: bool = True, api: int = CREW_PORTAL):
base_url = self.crew_portal_base_url if api == CREW_PORTAL else self.integration_base_url
def request(
self,
endpoint: str,
payload: dict,
add_token: bool = True,
wrap_request: bool = True,
extract_response: bool = True,
api: int = CREW_PORTAL,
):
base_url = (
self.crew_portal_base_url
if api == CREW_PORTAL
else self.integration_base_url
)
url = urljoin(base_url, endpoint)
req = Request(url)
@ -48,48 +71,62 @@ class Adonis:
res = res[endpoint + "Result"]
if not res["Authentication_Approved"]:
raise AdonisAuthenticationDeniedError(
res["Authentication_ReasonDenied"])
raise AdonisAuthenticationDeniedError(res["Authentication_ReasonDenied"])
if res["ErrorText"]:
raise AdonisError(res["ErrorText"])
return res
def requestIntegration(
self,
endpoint: str,
payload: dict,
add_token: bool = True,
wrap_request: bool = True,
extract_response: bool = True,
):
return self.request(
endpoint, payload, add_token, wrap_request, extract_response, INTEGRATION
)
def requestIntegration(self, endpoint: str, payload: dict, add_token: bool = True, wrap_request: bool = True, extract_response: bool = True):
return self.request(endpoint, payload, add_token, wrap_request, extract_response, INTEGRATION)
def requestCrewPortal(self, endpoint: str, payload: dict, add_token: bool = True, wrap_request: bool = True, extract_response: bool = True):
return self.request(endpoint, payload, add_token, wrap_request, extract_response, CREW_PORTAL)
def requestCrewPortal(
self,
endpoint: str,
payload: dict,
add_token: bool = True,
wrap_request: bool = True,
extract_response: bool = True,
):
return self.request(
endpoint, payload, add_token, wrap_request, extract_response, CREW_PORTAL
)
def getToken(self, lifetime: int = 60, api: int = CREW_PORTAL):
data = {
"credentials": {
"Login": self.login,
"Password": self.password,
"LifeTime": lifetime
"LifeTime": lifetime,
}
}
method = self.requestCrewPortal if api == CREW_PORTAL else self.requestIntegration
method = (
self.requestCrewPortal if api == CREW_PORTAL else self.requestIntegration
)
result = method("GNL_API_AUTHENTICATION", data, False, False)
return result["Authentication_Token"]
def getCrewPersonalData(self, pin):
data = {
"Pin": pin
}
data = {"Pin": pin}
result = self.requestCrewPortal("DG_GeneralDetailsRead", data)
return result
def getCrewProfilePicture(self, pin):
data = {
"Pin": pin
}
data = {"Pin": pin}
try:
result = self.requestCrewPortal("DG_ProfilePictureRead", data)
@ -117,8 +154,8 @@ class Adonis:
"Filter": filters,
"Pagination": pagination or "None",
"RowsByPage": paginate_by,
"Page": page
}
"Page": page,
}
try:
result = self.requestIntegration("GNL_APMCrewListViews", data)
@ -134,8 +171,10 @@ class Adonis:
pnum = 1
while True:
page = self.getCrewListView(view, pagination="True", paginate_by=str(paginate_by), page=str(pnum))
page = self.getCrewListView(
view, pagination="True", paginate_by=str(paginate_by), page=str(pnum)
)
for record in json.loads(page["Result"]):
requirements.append((record["PIN"], record["Document code"]))
@ -144,4 +183,82 @@ class Adonis:
pnum += 1
return requirements
return requirements
def getCodes(self):
try:
return self.requestIntegration("DG_GetCodes")
except AdonisError:
return None
def writeCompetence(
self,
pin,
competence_code,
certificate_number,
from_timestamp=None,
to_timestamp=None,
issued_by="pyadonis",
attachment=None,
**kwargs,
):
from_dt = (
datetime.datetime.fromtimestamp(from_timestamp)
if from_timestamp
else datetime.datetime.now()
)
from_dt = from_dt.replace(hour=0, minute=0, second=0, microsecond=0)
from_dt_string = from_dt.isoformat()
if to_timestamp:
to_dt = datetime.datetime.fromtimestamp(to_timestamp)
else:
codes = self.getCodes()
expiry_days = 0
for competence in codes["Competence"]:
if competence["Code"] == competence_code:
expiry_days = int(competence["DefaultExpiryDays"])
if expiry_days == 0:
to_dt_string = ""
else:
if not expiry_days % 365:
to_dt = from_dt + relativedelta(years=expiry_days / 365)
elif not expiry_days % 30:
to_dt = from_dt + relativedelta(months=expiry_days / 30)
else:
to_dt = from_dt + relativedelta(days=expiry_days)
to_dt_string = to_dt.isoformat()
payload = {
"Certificate": {
"Pin": pin,
"Code": competence_code,
"Number": certificate_number,
"DateFrom": from_dt_string,
"ExpiryDate": to_dt_string,
"IssuedBy": issued_by,
"Planned": "False",
}
}
if attachment:
payload["Certificate"]["FileName"] = "certificate.pdf"
if isinstance(attachment, bytes):
attachment = attachment.decode()
payload["Certificate"]["DocumentBase64"] = (
"data:application/pdf;" + attachment
)
payload.update(kwargs)
return self.requestCrewPortal("DG_CertificateCreate", payload)
def getCompetences(self, pin):
payload = {"Pin": pin}
return self.requestCrewPortal("DG_CertificateRead", payload)

1
src/pyadonis/classes/config.py
View file

@ -1,5 +1,6 @@
from configparser import ConfigParser
class Config:
def __init__(self, path):
self.config = ConfigParser()

8
src/pyadonis/classes/crew.py
View file

@ -22,8 +22,12 @@ class Crew:
@property
def email(self):
return self._personal_data["Emails"][0]["Email"] if self._personal_data["Emails"] else ""
return (
self._personal_data["Emails"][0]["Email"]
if self._personal_data["Emails"]
else ""
)
@property
def profilePicture(self):
return self._profile_picture
return self._profile_picture

3
src/pyadonis/classes/exceptions.py
View file

@ -1,5 +1,6 @@
class AdonisError(Exception):
pass
class AdonisAuthenticationDeniedError(AdonisError):
pass
pass

2
src/pyadonis/pindata.py
View file

@ -24,7 +24,7 @@ if __name__ == "__main__":
crewdata["middleName"] = crew.middleName
crewdata["lastName"] = crew.lastName
crewdata["dob"] = crew.birthDate
crewdata["password"] = crew.birthDate.replace("-","")
crewdata["password"] = crew.birthDate.replace("-", "")
crewdata["email"] = crew.email
crewdata["profilePicture"] = crew.profilePicture