Fork FedExTrack for DHLTrack
This commit is contained in:
parent
6a59140f5a
commit
444d4372ac
12 changed files with 99 additions and 141 deletions
|
@ -14,7 +14,7 @@ before_script:
|
|||
test:
|
||||
stage: test
|
||||
script:
|
||||
- echo "[FedEx]" > config.ini
|
||||
- echo "[DHL]" > config.ini
|
||||
- echo "key = ${API_KEY}" >> config.ini
|
||||
- echo "secret = ${API_SECRET}" >> config.ini
|
||||
- python -m unittest test.py
|
||||
|
|
2
LICENSE
2
LICENSE
|
@ -1,4 +1,4 @@
|
|||
Copyright (c) 2023 Kumi Mitterer <fedextrack@kumi.email>
|
||||
Copyright (c) 2023 Kumi Mitterer <dhltrack@kumi.email>
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
|
|
14
README.md
14
README.md
|
@ -1,25 +1,25 @@
|
|||
# FedEx Tracking API Python Client
|
||||
# DHL Tracking API Python Client
|
||||
|
||||
This is a Python client for the [FedEx Tracking API](https://developer.fedex.com/api/en-at/catalog/track/v1/docs.html).
|
||||
This is a Python client for the [DHL Shipment Tracking Unified API](https://developer.dhl.com/api-reference/shipment-tracking).
|
||||
|
||||
It is not fully featured yet, but it is a good starting point. It requires you to have a FedEx developer account and an API key.
|
||||
It is not fully featured yet, but it is a good starting point. It requires you to have a DHL developer account and an API key.
|
||||
|
||||
## Installation
|
||||
|
||||
```bash
|
||||
pip install fedextrack
|
||||
pip install dhltrack
|
||||
```
|
||||
|
||||
## Usage
|
||||
|
||||
```python
|
||||
from fedextrack import FedEx
|
||||
from dhltrack import DHL
|
||||
|
||||
api = FedEx("YOUR_API_KEY", "YOUR_API_SECRET")
|
||||
api = DHL("YOUR_API_KEY", "YOUR_API_SECRET")
|
||||
|
||||
# Realtime tracking
|
||||
|
||||
tracking = api.tracking("YOUR_SHIPMENT_NUMBER")
|
||||
tracking = api.track("YOUR_SHIPMENT_NUMBER")
|
||||
```
|
||||
|
||||
## License
|
||||
|
|
|
@ -1,3 +1,3 @@
|
|||
[FedEx]
|
||||
[DHL]
|
||||
key = api_key
|
||||
secret = api_secret
|
||||
|
|
|
@ -3,12 +3,12 @@ requires = ["hatchling"]
|
|||
build-backend = "hatchling.build"
|
||||
|
||||
[project]
|
||||
name = "fedextrack"
|
||||
name = "dhltrack"
|
||||
version = "0.9.0"
|
||||
authors = [
|
||||
{ name="Kumi Mitterer", email="fedextrack@kumi.email" },
|
||||
{ name="Kumi Mitterer", email="dhltrack@kumi.email" },
|
||||
]
|
||||
description = "Simple Python wrapper to fetch data from FedEx Tracking API"
|
||||
description = "Simple Python wrapper to fetch data from DHL Tracking API"
|
||||
readme = "README.md"
|
||||
license = { file="LICENSE" }
|
||||
requires-python = ">=3.10"
|
||||
|
@ -19,5 +19,5 @@ classifiers = [
|
|||
]
|
||||
|
||||
[project.urls]
|
||||
"Homepage" = "https://kumig.it/kumitterer/fedextrack"
|
||||
"Bug Tracker" = "https://kumig.it/kumitterer/fedextrack/issues"
|
||||
"Homepage" = "https://kumig.it/kumitterer/dhltrack"
|
||||
"Bug Tracker" = "https://kumig.it/kumitterer/dhltrack/issues"
|
|
@ -1,2 +1,2 @@
|
|||
from .http import HTTPRequest
|
||||
from .fedex import FedEx
|
||||
from .dhl import DHL
|
53
src/dhltrack/classes/dhl.py
Normal file
53
src/dhltrack/classes/dhl.py
Normal file
|
@ -0,0 +1,53 @@
|
|||
from hashlib import md5
|
||||
from configparser import ConfigParser
|
||||
from urllib.parse import urlencode
|
||||
from typing import Optional
|
||||
|
||||
import json
|
||||
|
||||
from .http import HTTPRequest
|
||||
|
||||
|
||||
class DHL:
|
||||
BASE_URL = "https://api-eu.dhl.com/track/"
|
||||
BASE_URL_SANDBOX = "https://api-test.dhl.com/track/"
|
||||
|
||||
def __init__(self, key: str, secret: str, base_url: str = BASE_URL):
|
||||
self.key = key
|
||||
self.secret = secret
|
||||
self.base_url = base_url
|
||||
|
||||
@classmethod
|
||||
def from_config(cls, config: ConfigParser | str, section: str = "DHL") -> "DHL":
|
||||
if isinstance(config, str):
|
||||
temp_config = ConfigParser()
|
||||
temp_config.read(config)
|
||||
config = temp_config
|
||||
|
||||
key = config.get(section, "key")
|
||||
secret = config.get(section, "secret")
|
||||
base_url = config.get(section, "base_url", fallback=cls.BASE_URL)
|
||||
|
||||
return cls(key, secret, base_url)
|
||||
|
||||
def get_request(self, endpoint: str, add_token: bool = True) -> HTTPRequest:
|
||||
url = self.base_url + endpoint
|
||||
|
||||
request = HTTPRequest(url)
|
||||
|
||||
if add_token:
|
||||
request.add_header("DHL-API-Key", self.key)
|
||||
|
||||
return request
|
||||
|
||||
def track(self, tracking_number: str, **kwargs) -> bytes:
|
||||
params = {
|
||||
"trackingNumber": tracking_number,
|
||||
}
|
||||
|
||||
for key in ("service", "requesterCountryCode", "originCountryCode", "language", "offset", "limit"):
|
||||
if key in kwargs:
|
||||
params[key] = kwargs[key]
|
||||
|
||||
request = self.get_request(f"shipments?{urlencode(params)}")
|
||||
return request.execute()
|
24
src/dhltrack/classes/http.py
Normal file
24
src/dhltrack/classes/http.py
Normal file
|
@ -0,0 +1,24 @@
|
|||
from urllib.request import Request, urlopen
|
||||
from urllib.error import HTTPError
|
||||
from io import BytesIO
|
||||
|
||||
import gzip
|
||||
import json
|
||||
|
||||
|
||||
class HTTPRequest(Request):
|
||||
USER_AGENT = "Mozilla/5.0 (compatible; DHLTrack/dev; +https://kumig.it/kumitterer/dhltrack)"
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self.add_header("User-Agent", self.USER_AGENT)
|
||||
|
||||
def execute(self, load_json: bool = True, *args, **kwargs):
|
||||
try:
|
||||
response = urlopen(self, *args, **kwargs).read()
|
||||
except HTTPError as e:
|
||||
print(e.read())
|
||||
raise
|
||||
if load_json:
|
||||
response = json.loads(response)
|
||||
return response
|
|
@ -1,74 +0,0 @@
|
|||
from hashlib import md5
|
||||
from configparser import ConfigParser
|
||||
from urllib.parse import urlencode
|
||||
|
||||
import json
|
||||
|
||||
from .http import HTTPRequest
|
||||
|
||||
|
||||
class FedEx:
|
||||
BASE_URL = "https://apis.fedex.com/"
|
||||
TRACK_BY_NUMBER = "track/v1/trackingnumbers"
|
||||
OAUTH_TOKEN = "oauth/token"
|
||||
|
||||
def __init__(self, key: str, secret: str, base_url: str = BASE_URL):
|
||||
self.key = key
|
||||
self.secret = secret
|
||||
self.base_url = base_url
|
||||
|
||||
def get_token(self):
|
||||
message = {
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": self.key,
|
||||
"client_secret": self.secret,
|
||||
}
|
||||
|
||||
request = self.get_request(self.OAUTH_TOKEN, {}, False)
|
||||
|
||||
request.add_header("Content-Type", "application/x-www-form-urlencoded")
|
||||
request.data = urlencode(message).encode("utf-8")
|
||||
|
||||
response = request.execute()
|
||||
return response["access_token"]
|
||||
|
||||
@classmethod
|
||||
def from_config(cls, config: ConfigParser | str, section: str = "FedEx") -> "FedEx":
|
||||
if isinstance(config, str):
|
||||
temp_config = ConfigParser()
|
||||
temp_config.read(config)
|
||||
config = temp_config
|
||||
|
||||
key = config.get(section, "key")
|
||||
secret = config.get(section, "secret")
|
||||
base_url = config.get(section, "base_url", fallback=cls.BASE_URL)
|
||||
|
||||
return cls(key, secret, base_url)
|
||||
|
||||
def get_request(self, endpoint: str, message: dict = {}, add_token: bool = True) -> HTTPRequest:
|
||||
url = self.base_url + endpoint
|
||||
|
||||
request = HTTPRequest(url)
|
||||
|
||||
if message:
|
||||
request.add_json_payload(message)
|
||||
|
||||
if add_token:
|
||||
request.add_header("Authorization", "Bearer " + self.get_token())
|
||||
|
||||
return request
|
||||
|
||||
def track_by_tracking_number(self, tracking_number: str, include_detailed_scans: bool = True) -> bytes:
|
||||
message = {
|
||||
"include_detailed_scans": include_detailed_scans,
|
||||
"trackingInfo": [
|
||||
{
|
||||
"trackingNumberInfo": {
|
||||
"trackingNumber": tracking_number,
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
request = self.get_request(self.TRACK_BY_NUMBER, message)
|
||||
return request.execute()
|
|
@ -1,37 +0,0 @@
|
|||
from urllib.request import Request, urlopen
|
||||
from urllib.error import HTTPError
|
||||
from io import BytesIO
|
||||
|
||||
import gzip
|
||||
import json
|
||||
|
||||
|
||||
class HTTPRequest(Request):
|
||||
USER_AGENT = "Mozilla/5.0 (compatible; FedExTrack/dev; +https://kumig.it/kumitterer/fedextrack)"
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self.add_header("User-Agent", self.USER_AGENT)
|
||||
|
||||
@staticmethod
|
||||
def read(response):
|
||||
if response.info().get('Content-Encoding') == 'gzip':
|
||||
buf = BytesIO(response.read())
|
||||
f = gzip.GzipFile(fileobj=buf)
|
||||
return f.read()
|
||||
|
||||
return response.read()
|
||||
|
||||
def execute(self, load_json: bool = True, *args, **kwargs):
|
||||
try:
|
||||
response = self.read(urlopen(self, *args, **kwargs))
|
||||
except HTTPError as e:
|
||||
print(self.read(e))
|
||||
raise
|
||||
if load_json:
|
||||
response = json.loads(response)
|
||||
return response
|
||||
|
||||
def add_json_payload(self, payload: dict):
|
||||
self.add_header("Content-Type", "application/json")
|
||||
self.data = json.dumps(payload).encode("utf-8")
|
20
test.py
20
test.py
|
@ -3,7 +3,7 @@ from configparser import ConfigParser
|
|||
|
||||
import json
|
||||
|
||||
from fedextrack import *
|
||||
from dhltrack import *
|
||||
|
||||
class TestHTTPRequest(TestCase):
|
||||
def test_http_request(self):
|
||||
|
@ -11,25 +11,17 @@ class TestHTTPRequest(TestCase):
|
|||
response = http.execute()
|
||||
self.assertEqual(response["headers"]["User-Agent"], http.USER_AGENT)
|
||||
|
||||
def test_http_request_with_json_payload(self):
|
||||
http = HTTPRequest("https://httpbin.org/post")
|
||||
http.add_json_payload({"foo": "bar"})
|
||||
response = http.execute()
|
||||
self.assertEqual(response["headers"]["User-Agent"], http.USER_AGENT)
|
||||
self.assertEqual(response["headers"]["Content-Type"], "application/json")
|
||||
self.assertEqual(response["json"]["foo"], "bar")
|
||||
|
||||
class TestFedEx(TestCase):
|
||||
class TestDHL(TestCase):
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self.config = ConfigParser()
|
||||
self.config.read("config.ini")
|
||||
self.fedex = FedEx.from_config(self.config)
|
||||
self.dhl = DHL.from_config(self.config)
|
||||
|
||||
def test_tracking(self):
|
||||
tracking_number = "702395541585"
|
||||
response = self.fedex.track_by_tracking_number(tracking_number)
|
||||
self.assertEqual(response["output"]["completeTrackResults"][0]["trackingNumber"], tracking_number)
|
||||
tracking_number = "LE650235858DE"
|
||||
response = self.dhl.track(tracking_number)
|
||||
self.assertEqual(response["shipments"][0]["id"], tracking_number)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
Loading…
Reference in a new issue