Last active
January 29, 2025 09:58
-
-
Save CheeseCake87/1bf987d157a1de0763ebc13eed4e769b to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| """ | |
| requirements: | |
| pip install pytz | |
| pip install python-dotenv | |
| pip install requests | |
| - create .env file with the following - | |
| DVLA_VES_API_KEY="" | |
| DVLA_MOTH_TENANT_ID="" | |
| DVLA_MOTH_CLIENT_ID="" | |
| DVLA_MOTH_CLIENT_SECRET="" | |
| DVLA_MOTH_API_KEY="" | |
| -- | |
| Vehicle Enquiry Service (VES) API Guide | |
| https://developer-portal.driver-vehicle-licensing.api.gov.uk/apis/vehicle-enquiry-service/vehicle-enquiry-service-description.html#vehicle-enquiry-service-ves-api-guide | |
| MOT History API Guide | |
| https://documentation.history.mot.api.gov.uk/ | |
| """ | |
| import json | |
| import os | |
| from datetime import datetime | |
| from datetime import timedelta | |
| from pathlib import Path | |
| import requests | |
| from dotenv import load_dotenv | |
| from pytz import timezone | |
| from pprint import pprint | |
| class DatetimeDelta: | |
| """ | |
| Produces timezone-aware dates | |
| Returns self when a method is called | |
| https://en.wikipedia.org/wiki/List_of_tz_database_time_zones#List | |
| """ | |
| _local_tz: str | |
| _format: str | |
| _timezone: timezone | |
| _datetime: datetime | |
| def __init__( | |
| self, | |
| ltz: str = "Europe/London", | |
| format_: str = "%Y-%m-%d %H:%M:%S", | |
| datetime_: datetime = None, | |
| ): | |
| self._local_tz = ltz | |
| self._format = format_ | |
| self._timezone = timezone(ltz) | |
| if datetime_: | |
| self._datetime = datetime_.replace(tzinfo=self._timezone) | |
| else: | |
| self._datetime = datetime.now(self._timezone) | |
| def set_format(self, format_: str = "%Y-%m-%d %H:%M:%S") -> "DatetimeDelta": | |
| self._format = format_ | |
| return self | |
| def days(self, days_delta: int) -> "DatetimeDelta": | |
| self._datetime = self._datetime + timedelta(days=days_delta) | |
| return self | |
| def hours(self, hours_delta: int) -> "DatetimeDelta": | |
| self._datetime = self._datetime + timedelta(hours=hours_delta) | |
| return self | |
| def minutes(self, minuets_delta: int) -> "DatetimeDelta": | |
| self._datetime = self._datetime + timedelta(minutes=minuets_delta) | |
| return self | |
| def __str__(self) -> str: | |
| return self._datetime.strftime(self._format) | |
| @property | |
| def datetime(self) -> datetime: | |
| return self._datetime | |
| @property | |
| def timestamp(self) -> int: | |
| return int(self._datetime.timestamp()) | |
| @property | |
| def timezone(self) -> timezone: | |
| return self._timezone | |
| class DVLAException(Exception): | |
| pass | |
| class DVLAVehicleEnquiry: | |
| _api_url: str = "https://driver-vehicle-licensing.api.gov.uk/vehicle-enquiry/v1/vehicles" | |
| api_key: str | |
| def __init__(self, api_key: str): | |
| self.api_key = api_key | |
| @staticmethod | |
| def _clean(string: str): | |
| return string.replace(" ", "").upper() | |
| def registration_lookup(self, registration: str): | |
| res = requests.post( | |
| self._api_url, | |
| headers={ | |
| "X-API-Key": self.api_key | |
| }, | |
| json={ | |
| "registrationNumber": self._clean(registration) | |
| } | |
| ) | |
| try: | |
| return res.json() | |
| except json.decoder.JSONDecodeError: | |
| return {"error": "No data found."} | |
| class DVLAMOTHistory: | |
| working_dir: Path | |
| token_json_file: Path | |
| _api_url: str = "https://history.mot.api.gov.uk" | |
| _auth_url: str = "https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token" | |
| _scope: str = "https://tapi.dvsa.gov.uk/.default" | |
| _refresh_auth_url: str = "/v1/trade/credentials" | |
| _registration_url: str = "/v1/trade/vehicles/registration/{registration}" | |
| _vin_url: str = "/v1/trade/vehicles/vin/{vin}" | |
| tenant_id: str | |
| client_id: str | |
| client_secret: str | |
| api_key: str | |
| def __init__(self, tenant_id: str, client_id: str, client_secret: str, api_key: str): | |
| self.tenant_id = tenant_id | |
| self.client_id = client_id | |
| self.client_secret = client_secret | |
| self.api_key = api_key | |
| self._auth_url = self._auth_url.format(tenant_id=tenant_id) | |
| self.working_dir = Path(__file__).parent.resolve() | |
| self.token_json_file = self.working_dir / "token.json" | |
| if not self.token_json_file.exists(): | |
| self._create_token_file() | |
| @staticmethod | |
| def _clean(string: str): | |
| return string.replace(" ", "").upper() | |
| def _fetch_token(self) -> str | None: | |
| res = requests.post( | |
| self._auth_url, | |
| headers={ | |
| "Content-Type": "application/x-www-form-urlencoded", | |
| }, | |
| data={ | |
| "grant_type": "client_credentials", | |
| "scope": self._scope, | |
| "client_id": self.client_id, | |
| "client_secret": self.client_secret | |
| } | |
| ) | |
| try: | |
| jsond = res.json() | |
| except json.decoder.JSONDecodeError: | |
| jsond = {"error": "No data found."} | |
| if "access_token" in jsond: | |
| return jsond["access_token"] | |
| return None | |
| def _create_token_file(self): | |
| expired_date = DatetimeDelta().minutes(-1).timestamp | |
| return self.token_json_file.write_text(json.dumps({ | |
| "expires": expired_date, | |
| "token": '-', | |
| })) | |
| def _load_token_file(self): | |
| return json.loads(self.token_json_file.read_text()) | |
| def _save_token_file(self, expires: int, token: str): | |
| self.token_json_file.write_text(json.dumps({ | |
| "expires": expires, | |
| "token": token, | |
| })) | |
| @property | |
| def token(self): | |
| token = self._load_token_file() | |
| now = DatetimeDelta().timestamp | |
| expires = token["expires"] | |
| token = token["token"] | |
| if expires < now: # Check if expires is a timestamp before now | |
| if token := self._fetch_token(): # fetch and store new token | |
| new_expires = DatetimeDelta().minutes(55).timestamp # set new time in future | |
| self._save_token_file(new_expires, token) # update previous token | |
| return token | |
| else: | |
| raise DVLAException("Token not found.") | |
| else: # token has not expired, send stored | |
| return token | |
| def registration_lookup(self, registration: str): | |
| token = self.token | |
| res = requests.get( | |
| self._api_url + self._registration_url.format(registration=self._clean(registration)), | |
| headers={ | |
| "Authorization": f"Bearer {token}", | |
| "X-API-Key": self.api_key | |
| } | |
| ) | |
| try: | |
| return res.json() | |
| except json.decoder.JSONDecodeError: | |
| return {"error": "No data found."} | |
| def vin_lookup(self, vin: str): | |
| token = self.token | |
| res = requests.get( | |
| self._api_url + self._vin_url.format(vin=self._clean(vin)), | |
| headers={ | |
| "Authorization": f"Bearer {token}", | |
| "X-API-Key": self.api_key | |
| } | |
| ) | |
| try: | |
| return res.json() | |
| except json.decoder.JSONDecodeError: | |
| return {"error": "No data found."} | |
| if __name__ == '__main__': | |
| load_dotenv() | |
| mot_history = DVLAMOTHistory( | |
| tenant_id=os.getenv("DVLA_MOTH_TENANT_ID"), | |
| client_id=os.getenv("DVLA_MOTH_CLIENT_ID"), | |
| client_secret=os.getenv("DVLA_MOTH_CLIENT_SECRET"), | |
| api_key=os.getenv("DVLA_MOTH_API_KEY") | |
| ) | |
| vehicle_enquiry = DVLAVehicleEnquiry(api_key=os.getenv("DVLA_VES_API_KEY")) | |
| pprint(mot_history.registration_lookup("REGHERE")) | |
| pprint(vehicle_enquiry.registration_lookup("REGHERE")) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment