From 1a98f08a3ce24d3482cf48dcb4ef7f397a037804 Mon Sep 17 00:00:00 2001 From: Senthur Date: Mon, 7 Oct 2024 15:34:20 -0400 Subject: [PATCH] Refactored client class. --- onshape_api/__init__.py | 14 + onshape_api/client.py | 234 ---------------- onshape_api/connection.py | 438 ++++++++++++++++++++++++++++++ onshape_api/examples/__init__.py | 0 onshape_api/examples/bike/main.py | 25 ++ onshape_api/main.py | 26 -- onshape_api/onshape.py | 248 ----------------- onshape_api/utilities.py | 289 ++++++++++++++++++++ onshape_api/utils.py | 64 ----- pyproject.toml | 1 + 10 files changed, 767 insertions(+), 572 deletions(-) delete mode 100644 onshape_api/client.py create mode 100644 onshape_api/connection.py create mode 100644 onshape_api/examples/__init__.py create mode 100644 onshape_api/examples/bike/main.py delete mode 100644 onshape_api/main.py delete mode 100644 onshape_api/onshape.py create mode 100644 onshape_api/utilities.py delete mode 100644 onshape_api/utils.py diff --git a/onshape_api/__init__.py b/onshape_api/__init__.py index e69de29..0ce51bb 100644 --- a/onshape_api/__init__.py +++ b/onshape_api/__init__.py @@ -0,0 +1,14 @@ +from importlib import metadata as importlib_metadata + + +def get_version() -> str: + try: + return importlib_metadata.version(__name__) + except importlib_metadata.PackageNotFoundError: # pragma: no cover + return "unknown" + + +__version__: str = get_version() + +from onshape_api.connection import * # noqa: F403 E402 +from onshape_api.utilities import * # noqa: F403 E402 diff --git a/onshape_api/client.py b/onshape_api/client.py deleted file mode 100644 index a578444..0000000 --- a/onshape_api/client.py +++ /dev/null @@ -1,234 +0,0 @@ -""" -client -====== - -Convenience functions for working with the Onshape API -""" - -import mimetypes -import os -import secrets -import string - -from onshape_api.onshape import Onshape - - -class Client: - """ - Defines methods for testing the Onshape API. Comes with several methods: - - - Create a document - - Delete a document - - Get a list of documents - - Attributes: - - stack (str, default='https://cad.onshape.com'): Base URL - - logging (bool, default=True): Turn logging on or off - """ - - def __init__(self, stack="https://cad.onshape.com", logging=True): - """ - Instantiates a new Onshape client. - - Args: - - stack (str, default='https://cad.onshape.com'): Base URL - - logging (bool, default=True): Turn logging on or off - """ - - self._stack = stack - self._api = Onshape(stack=stack, logging=logging) - - def new_document(self, name="Test Document", owner_type=0, public=False): - """ - Create a new document. - - Args: - - name (str, default='Test Document'): The doc name - - owner_type (int, default=0): 0 for user, 1 for company, 2 for team - - public (bool, default=False): Whether or not to make doc public - - Returns: - - requests.Response: Onshape response data - """ - - payload = {"name": name, "ownerType": owner_type, "isPublic": public} - - return self._api.request("post", "/api/documents", body=payload) - - def rename_document(self, did, name): - """ - Renames the specified document. - - Args: - - did (str): Document ID - - name (str): New document name - - Returns: - - requests.Response: Onshape response data - """ - - payload = {"name": name} - - return self._api.request("post", "/api/documents/" + did, body=payload) - - def del_document(self, did): - """ - Delete the specified document. - - Args: - - did (str): Document ID - - Returns: - - requests.Response: Onshape response data - """ - - return self._api.request("delete", "/api/documents/" + did) - - def get_document(self, did): - """ - Get details for a specified document. - - Args: - - did (str): Document ID - - Returns: - - requests.Response: Onshape response data - """ - - return self._api.request("get", "/api/documents/" + did) - - def list_documents(self): - """ - Get list of documents for current user. - - Returns: - - requests.Response: Onshape response data - """ - - return self._api.request("get", "/api/documents") - - def create_assembly(self, did, wid, name="My Assembly"): - """ - Creates a new assembly element in the specified document / workspace. - - Args: - - did (str): Document ID - - wid (str): Workspace ID - - name (str, default='My Assembly') - - Returns: - - requests.Response: Onshape response data - """ - - payload = {"name": name} - - return self._api.request("post", "/api/assemblies/d/" + did + "/w/" + wid, body=payload) - - def get_features(self, did, wid, eid): - """ - Gets the feature list for specified document / workspace / part studio. - - Args: - - did (str): Document ID - - wid (str): Workspace ID - - eid (str): Element ID - - Returns: - - requests.Response: Onshape response data - """ - - return self._api.request("get", "/api/partstudios/d/" + did + "/w/" + wid + "/e/" + eid + "/features") - - def get_partstudio_tessellatededges(self, did, wid, eid): - """ - Gets the tessellation of the edges of all parts in a part studio. - - Args: - - did (str): Document ID - - wid (str): Workspace ID - - eid (str): Element ID - - Returns: - - requests.Response: Onshape response data - """ - - return self._api.request( - "get", - "/api/partstudios/d/" + did + "/w/" + wid + "/e/" + eid + "/tessellatededges", - ) - - def upload_blob(self, did, wid, filepath="./blob.json"): - """ - Uploads a file to a new blob element in the specified doc. - - Args: - - did (str): Document ID - - wid (str): Workspace ID - - filepath (str, default='./blob.json'): Blob element location - - Returns: - - requests.Response: Onshape response data - """ - - chars = string.ascii_letters + string.digits - boundary_key = "".join(secrets.choice(chars) for i in range(8)) - - mimetype = mimetypes.guess_type(filepath)[0] - encoded_filename = os.path.basename(filepath) - file_content_length = str(os.path.getsize(filepath)) - blob = open(filepath) - - req_headers = {"Content-Type": f'multipart/form-data; boundary="{boundary_key}"'} - - # build request body - payload = ( - "--" - + boundary_key - + '\r\nContent-Disposition: form-data; name="encodedFilename"\r\n\r\n' - + encoded_filename - + "\r\n" - ) - payload += ( - "--" - + boundary_key - + '\r\nContent-Disposition: form-data; name="fileContentLength"\r\n\r\n' - + file_content_length - + "\r\n" - ) - payload += ( - "--" - + boundary_key - + '\r\nContent-Disposition: form-data; name="file"; filename="' - + encoded_filename - + '"\r\n' - ) - payload += "Content-Type: " + mimetype + "\r\n\r\n" - payload += blob.read() - payload += "\r\n--" + boundary_key + "--" - - return self._api.request( - "post", - "/api/blobelements/d/" + did + "/w/" + wid, - headers=req_headers, - body=payload, - ) - - def part_studio_stl(self, did, wid, eid): - """ - Exports STL export from a part studio - - Args: - - did (str): Document ID - - wid (str): Workspace ID - - eid (str): Element ID - - Returns: - - requests.Response: Onshape response data - """ - - req_headers = {"Accept": "application/vnd.onshape.v1+octet-stream"} - return self._api.request( - "get", - "/api/partstudios/d/" + did + "/w/" + wid + "/e/" + eid + "/stl", - headers=req_headers, - ) diff --git a/onshape_api/connection.py b/onshape_api/connection.py new file mode 100644 index 0000000..93b271b --- /dev/null +++ b/onshape_api/connection.py @@ -0,0 +1,438 @@ +import base64 +import datetime +import hashlib +import hmac +import json +import mimetypes +import os +import secrets +import string +from urllib.parse import parse_qs, urlencode, urlparse + +import requests +from dotenv import load_dotenv + +from onshape_api.utilities import LOG_LEVEL, LOGGER + +__all__ = ["Client"] + +""" +client +====== + +Convenience functions for working with the Onshape API +""" + +# TODO: Add asyncio support for async requests + + +def load_env_variables(env): + """ + Load environment variables from the specified .env file. + + Args: + env (str): Path to the .env file. + + Returns: + tuple: A tuple containing the URL, ACCESS_KEY, and SECRET_KEY. + + Raises: + FileNotFoundError: If the .env file does not exist. + ValueError: If any of the required environment variables are missing. + """ + if not os.path.isfile(env): + raise FileNotFoundError(f"{env} file not found") + + load_dotenv(env) + + url = os.getenv("URL") + access_key = os.getenv("ACCESS_KEY") + secret_key = os.getenv("SECRET_KEY") + + if not url or not access_key or not secret_key: + missing_vars = [var for var in ["URL", "ACCESS_KEY", "SECRET_KEY"] if not os.getenv(var)] + raise ValueError(f"Missing required environment variables: {', '.join(missing_vars)}") + + return url, access_key, secret_key + + +def make_nonce(): + """ + Generate a unique ID for the request, 25 chars in length + + Returns: + - str: Cryptographic nonce + """ + + chars = string.digits + string.ascii_letters + nonce = "".join(secrets.choice(chars) for i in range(25)) + LOGGER.debug(f"nonce created: {nonce}") + + return nonce + + +class Client: + """ + Provides access to the Onshape REST API. + + Attributes: + - env (str, default='./.env'): Location of the environment file + - logging (bool, default=True): Turn logging on or off + """ + + def __init__(self, env="./.env", log_file="./onshape.log", log_level=1): + """ + Instantiates an instance of the Onshape class. Reads credentials from a .env file. + + The .env file should be stored in the root project folder; optionally, + you can specify the location of a different file. + + Args: + - env (str, default='./.env'): Environment file location + """ + + self._url, self._access_key, self._secret_key = load_env_variables(env) + LOGGER.set_file_name(log_file) + LOGGER.set_stream_level(LOG_LEVEL[log_level]) + LOGGER.info(f"Onshape API initialized with env file: {env}") + + def new_document(self, name="Test Document", owner_type=0, public=False): + """ + Create a new document. + + Args: + - name (str, default='Test Document'): The doc name + - owner_type (int, default=0): 0 for user, 1 for company, 2 for team + - public (bool, default=False): Whether or not to make doc public + + Returns: + - requests.Response: Onshape response data + """ + + payload = {"name": name, "ownerType": owner_type, "isPublic": public} + + return self.request("post", "/api/documents", body=payload) + + def rename_document(self, did, name): + """ + Renames the specified document. + + Args: + - did (str): Document ID + - name (str): New document name + + Returns: + - requests.Response: Onshape response data + """ + + payload = {"name": name} + + return self.request("post", "/api/documents/" + did, body=payload) + + def del_document(self, did): + """ + Delete the specified document. + + Args: + - did (str): Document ID + + Returns: + - requests.Response: Onshape response data + """ + + return self.request("delete", "/api/documents/" + did) + + def get_document(self, did): + """ + Get details for a specified document. + + Args: + - did (str): Document ID + + Returns: + - requests.Response: Onshape response data + """ + + return self.request("get", "/api/documents/" + did) + + def list_documents(self): + """ + Get list of documents for current user. + + Returns: + - requests.Response: Onshape response data + """ + + return self.request("get", "/api/documents") + + def create_assembly(self, did, wid, name="My Assembly"): + """ + Creates a new assembly element in the specified document / workspace. + + Args: + - did (str): Document ID + - wid (str): Workspace ID + - name (str, default='My Assembly') + + Returns: + - requests.Response: Onshape response data + """ + + payload = {"name": name} + + return self.request("post", "/api/assemblies/d/" + did + "/w/" + wid, body=payload) + + def get_features(self, did, wid, eid): + """ + Gets the feature list for specified document / workspace / part studio. + + Args: + - did (str): Document ID + - wid (str): Workspace ID + - eid (str): Element ID + + Returns: + - requests.Response: Onshape response data + """ + + return self.request("get", "/api/partstudios/d/" + did + "/w/" + wid + "/e/" + eid + "/features") + + def get_partstudio_tessellatededges(self, did, wid, eid): + """ + Gets the tessellation of the edges of all parts in a part studio. + + Args: + - did (str): Document ID + - wid (str): Workspace ID + - eid (str): Element ID + + Returns: + - requests.Response: Onshape response data + """ + + return self.request( + "get", + "/api/partstudios/d/" + did + "/w/" + wid + "/e/" + eid + "/tessellatededges", + ) + + def upload_blob(self, did, wid, filepath="./blob.json"): + """ + Uploads a file to a new blob element in the specified doc. + + Args: + - did (str): Document ID + - wid (str): Workspace ID + - filepath (str, default='./blob.json'): Blob element location + + Returns: + - requests.Response: Onshape response data + """ + + chars = string.ascii_letters + string.digits + boundary_key = "".join(secrets.choice(chars) for i in range(8)) + + mimetype = mimetypes.guess_type(filepath)[0] + encoded_filename = os.path.basename(filepath) + file_content_length = str(os.path.getsize(filepath)) + blob = open(filepath) + + req_headers = {"Content-Type": f'multipart/form-data; boundary="{boundary_key}"'} + + # build request body + payload = ( + "--" + + boundary_key + + '\r\nContent-Disposition: form-data; name="encodedFilename"\r\n\r\n' + + encoded_filename + + "\r\n" + ) + payload += ( + "--" + + boundary_key + + '\r\nContent-Disposition: form-data; name="fileContentLength"\r\n\r\n' + + file_content_length + + "\r\n" + ) + payload += ( + "--" + + boundary_key + + '\r\nContent-Disposition: form-data; name="file"; filename="' + + encoded_filename + + '"\r\n' + ) + payload += "Content-Type: " + mimetype + "\r\n\r\n" + payload += blob.read() + payload += "\r\n--" + boundary_key + "--" + + return self.request( + "post", + "/api/blobelements/d/" + did + "/w/" + wid, + headers=req_headers, + body=payload, + ) + + def part_studio_stl(self, did, wid, eid): + """ + Exports STL export from a part studio + + Args: + - did (str): Document ID + - wid (str): Workspace ID + - eid (str): Element ID + + Returns: + - requests.Response: Onshape response data + """ + + req_headers = {"Accept": "application/vnd.onshape.v1+octet-stream"} + return self.request( + "get", + "/api/partstudios/d/" + did + "/w/" + wid + "/e/" + eid + "/stl", + headers=req_headers, + ) + + def _make_auth(self, method, date, nonce, path, query=None, ctype="application/json"): + """ + Create the request signature to authenticate + + Args: + - method (str): HTTP method + - date (str): HTTP date header string + - nonce (str): Cryptographic nonce + - path (str): URL pathname + - query (dict, default={}): URL query string in key-value pairs + - ctype (str, default='application/json'): HTTP Content-Type + """ + + if query is None: + query = {} + query = urlencode(query) + + hmac_str = ( + (method + "\n" + nonce + "\n" + date + "\n" + ctype + "\n" + path + "\n" + query + "\n") + .lower() + .encode("utf-8") + ) + + signature = base64.b64encode( + hmac.new(self._secret_key.encode("utf-8"), hmac_str, digestmod=hashlib.sha256).digest() + ) + auth = "On " + self._access_key + ":HmacSHA256:" + signature.decode("utf-8") + + LOGGER.debug(f"query: {query}, hmac_str: {hmac_str}, signature: {signature}, auth: {auth}") + + return auth + + def _make_headers(self, method, path, query=None, headers=None): + """ + Creates a headers object to sign the request + + Args: + - method (str): HTTP method + - path (str): Request path, e.g. /api/documents. No query string + - query (dict, default={}): Query string in key-value format + - headers (dict, default={}): Other headers to pass in + + Returns: + - dict: Dictionary containing all headers + """ + + if headers is None: + headers = {} + if query is None: + query = {} + date = datetime.datetime.utcnow().strftime("%a, %d %b %Y %H:%M:%S GMT") + nonce = make_nonce() + ctype = headers.get("Content-Type") if headers.get("Content-Type") else "application/json" + + auth = self._make_auth(method, date, nonce, path, query=query, ctype=ctype) + + req_headers = { + "Content-Type": "application/json", + "Date": date, + "On-Nonce": nonce, + "Authorization": auth, + "User-Agent": "Onshape Python Sample App", + "Accept": "application/json", + } + + # add in user-defined headers + for h in headers: + req_headers[h] = headers[h] + + return req_headers + + def request(self, method, path, query=None, headers=None, body=None, base_url=None): + """ + Issues a request to Onshape + + Args: + - method (str): HTTP method + - path (str): Path e.g. /api/documents/:id + - query (dict, default={}): Query params in key-value pairs + - headers (dict, default={}): Key-value pairs of headers + - body (dict, default={}): Body for POST request + - base_url (str, default=None): Host, including scheme and port (if different from keys file) + + Returns: + - requests.Response: Object containing the response from Onshape + """ + body = body or {} + headers = headers or {} + query = query or {} + base_url = base_url or self._url + + req_headers = self._make_headers(method, path, query, headers) + url = self._build_url(base_url, path, query) + + LOGGER.debug(f"{body}") + LOGGER.debug(f"{req_headers}") + LOGGER.debug(f"request url: {url}") + + body = json.dumps(body) if isinstance(body, dict) else body + + res = self._send_request(method, url, req_headers, body) + + if res.status_code == 307: + return self._handle_redirect(res, method, headers) + else: + self._log_response(res) + + return res + + def _build_url(self, base_url, path, query): + return base_url + path + "?" + urlencode(query) + + def _send_request(self, method, url, headers, body): + return requests.request( + method, + url, + headers=headers, + data=body, + allow_redirects=False, + stream=True, + timeout=10, # Specify an appropriate timeout value in seconds + ) + + def _handle_redirect(self, res, method, headers): + location = urlparse(res.headers["Location"]) + querystring = parse_qs(location.query) + + LOGGER.debug(f"Request redirected to: {location.geturl()}") + + new_query = {key: querystring[key][0] for key in querystring} + new_base_url = location.scheme + "://" + location.netloc + + return self.request( + method, + location.path, + query=new_query, + headers=headers, + base_url=new_base_url, + ) + + def _log_response(self, res): + if not 200 <= res.status_code <= 206: + LOGGER.debug(f"Request failed, details: {res.text}") + else: + LOGGER.debug(f"Request succeeded, details: {res.text}") diff --git a/onshape_api/examples/__init__.py b/onshape_api/examples/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/onshape_api/examples/bike/main.py b/onshape_api/examples/bike/main.py new file mode 100644 index 0000000..adbb357 --- /dev/null +++ b/onshape_api/examples/bike/main.py @@ -0,0 +1,25 @@ +import onshape_api + +client = onshape_api.Client(env="/Users/holycow/Projects/onshape-api/.env") +print(client.list_documents().json()) + +# new_doc = c.new_document(name="Hello World", public=True).json() +# did = new_doc["id"] +# wid = new_doc["defaultWorkspace"]["id"] + +# details = c.get_document(did) +# print(details.json()) + +# asm = c.create_assembly(did, wid, name="Test Assembly") + +# if asm.json()["name"] == "Test Assembly": +# print("Assembly created") +# else: +# print("Error: Assembly not created") + +# c.del_document(did) +# trashed_doc = c.get_document(did) +# if trashed_doc.json()["trash"] is True: +# print("Document now in trash") +# else: +# print("Error: Document not trashed") diff --git a/onshape_api/main.py b/onshape_api/main.py deleted file mode 100644 index 7daa401..0000000 --- a/onshape_api/main.py +++ /dev/null @@ -1,26 +0,0 @@ -from onshape_api.client import Client - -stacks = {"cad": "https://cad.onshape.com"} - -c = Client(stack=stacks["cad"], logging=True) - -new_doc = c.new_document(name="Hello World", public=True).json() -did = new_doc["id"] -wid = new_doc["defaultWorkspace"]["id"] - -details = c.get_document(did) -print(details.json()) - -asm = c.create_assembly(did, wid, name="Test Assembly") - -if asm.json()["name"] == "Test Assembly": - print("Assembly created") -else: - print("Error: Assembly not created") - -# c.del_document(did) -# trashed_doc = c.get_document(did) -# if trashed_doc.json()["trash"] is True: -# print("Document now in trash") -# else: -# print("Error: Document not trashed") diff --git a/onshape_api/onshape.py b/onshape_api/onshape.py deleted file mode 100644 index f159732..0000000 --- a/onshape_api/onshape.py +++ /dev/null @@ -1,248 +0,0 @@ -""" -onshape -====== - -Provides access to the Onshape REST API -""" - -import base64 -import datetime -import hashlib -import hmac -import json -import os -import secrets -import string -from urllib.parse import parse_qs, urlencode, urlparse - -import requests - -import onshape_api.utils as utils - -__all__ = ["Onshape"] - - -class Onshape: - """ - Provides access to the Onshape REST API. - - Attributes: - - stack (str): Base URL - - keys (str, default='./keys.json'): Credentials location - - logging (bool, default=True): Turn logging on or off - """ - - def __init__(self, stack, keys="./keys.json", logging=True): - """ - Instantiates an instance of the Onshape class. Reads credentials from a JSON file - of this format: - - { - "http://cad.onshape.com": { - "access_key": "YOUR KEY HERE", - "secret_key": "YOUR KEY HERE" - }, - etc... add new object for each stack to test on - } - - The keys.json file should be stored in the root project folder; optionally, - you can specify the location of a different file. - - Args: - - stack (str): Base URL - - keys (str, default='./keys.json'): Credentials location - """ - - if not os.path.isfile(keys): - raise OSError(f"{keys} is not a file") - - with open(keys) as f: - try: - stacks = json.load(f) - if stack in stacks: - self._url = stack - self._access_key = stacks[stack]["access_key"] - self._secret_key = stacks[stack]["secret_key"] - self._logging = logging - else: - raise ValueError("specified stack not in file") - except TypeError as err: - raise ValueError(f"{keys} is not valid json") from err - - if self._logging: - utils.log(f"onshape instance created: url = {self._url}, access key = {self._access_key}") - - def _make_nonce(self): - """ - Generate a unique ID for the request, 25 chars in length - - Returns: - - str: Cryptographic nonce - """ - - chars = string.digits + string.ascii_letters - nonce = "".join(secrets.choice(chars) for i in range(25)) - - if self._logging: - utils.log(f"nonce created: {nonce}") - - return nonce - - def _make_auth(self, method, date, nonce, path, query=None, ctype="application/json"): - """ - Create the request signature to authenticate - - Args: - - method (str): HTTP method - - date (str): HTTP date header string - - nonce (str): Cryptographic nonce - - path (str): URL pathname - - query (dict, default={}): URL query string in key-value pairs - - ctype (str, default='application/json'): HTTP Content-Type - """ - - if query is None: - query = {} - query = urlencode(query) - - hmac_str = ( - (method + "\n" + nonce + "\n" + date + "\n" + ctype + "\n" + path + "\n" + query + "\n") - .lower() - .encode("utf-8") - ) - - signature = base64.b64encode( - hmac.new(self._secret_key.encode("utf-8"), hmac_str, digestmod=hashlib.sha256).digest() - ) - auth = "On " + self._access_key + ":HmacSHA256:" + signature.decode("utf-8") - - if self._logging: - utils.log({ - "query": query, - "hmac_str": hmac_str, - "signature": signature, - "auth": auth, - }) - - return auth - - def _make_headers(self, method, path, query=None, headers=None): - """ - Creates a headers object to sign the request - - Args: - - method (str): HTTP method - - path (str): Request path, e.g. /api/documents. No query string - - query (dict, default={}): Query string in key-value format - - headers (dict, default={}): Other headers to pass in - - Returns: - - dict: Dictionary containing all headers - """ - - if headers is None: - headers = {} - if query is None: - query = {} - date = datetime.datetime.utcnow().strftime("%a, %d %b %Y %H:%M:%S GMT") - nonce = self._make_nonce() - ctype = headers.get("Content-Type") if headers.get("Content-Type") else "application/json" - - auth = self._make_auth(method, date, nonce, path, query=query, ctype=ctype) - - req_headers = { - "Content-Type": "application/json", - "Date": date, - "On-Nonce": nonce, - "Authorization": auth, - "User-Agent": "Onshape Python Sample App", - "Accept": "application/json", - } - - # add in user-defined headers - for h in headers: - req_headers[h] = headers[h] - - return req_headers - - def request(self, method, path, query=None, headers=None, body=None, base_url=None): - """ - Issues a request to Onshape - - Args: - - method (str): HTTP method - - path (str): Path e.g. /api/documents/:id - - query (dict, default={}): Query params in key-value pairs - - headers (dict, default={}): Key-value pairs of headers - - body (dict, default={}): Body for POST request - - base_url (str, default=None): Host, including scheme and port (if different from keys file) - - Returns: - - requests.Response: Object containing the response from Onshape - """ - body = body or {} - headers = headers or {} - query = query or {} - base_url = base_url or self._url - - req_headers = self._make_headers(method, path, query, headers) - url = self._build_url(base_url, path, query) - - if self._logging: - self._log_request_details(body, req_headers, url) - - body = json.dumps(body) if isinstance(body, dict) else body - - res = self._send_request(method, url, req_headers, body) - - if res.status_code == 307: - return self._handle_redirect(res, method, headers) - else: - self._log_response(res) - - return res - - def _build_url(self, base_url, path, query): - return base_url + path + "?" + urlencode(query) - - def _log_request_details(self, body, req_headers, url): - utils.log(body) - utils.log(req_headers) - utils.log("request url: " + url) - - def _send_request(self, method, url, headers, body): - return requests.request( - method, - url, - headers=headers, - data=body, - allow_redirects=False, - stream=True, - timeout=10, # Specify an appropriate timeout value in seconds - ) - - def _handle_redirect(self, res, method, headers): - location = urlparse(res.headers["Location"]) - querystring = parse_qs(location.query) - - if self._logging: - utils.log("request redirected to: " + location.geturl()) - - new_query = {key: querystring[key][0] for key in querystring} - new_base_url = location.scheme + "://" + location.netloc - - return self.request( - method, - location.path, - query=new_query, - headers=headers, - base_url=new_base_url, - ) - - def _log_response(self, res): - if not 200 <= res.status_code <= 206: - if self._logging: - utils.log("request failed, details: " + res.text, level=1) - else: - if self._logging: - utils.log("request succeeded, details: " + res.text) diff --git a/onshape_api/utilities.py b/onshape_api/utilities.py new file mode 100644 index 0000000..fec0c41 --- /dev/null +++ b/onshape_api/utilities.py @@ -0,0 +1,289 @@ +""" +Logging module for opensourceleg library. + +Module Overview: + +This module defines a custom logger class, `Logger`, designed to log attributes +from class instances to a CSV file. It extends the `logging.Logger` class. + +Key Classes: + +- `LogLevel`: Enum class that defines the log levels supported by the `Logger` class. +- `Logger`: Logs attributes of class instances to a CSV file. It supports +setting different logging levels for file and stream handlers. +- `LOGGER`: Global instance of the `Logger` class that can be used throughout + +Usage Guide: + +1. Create an instance of the `Logger` class. +2. Optionally, set the logging levels for file and stream handlers using + `set_file_level` and `set_stream_level` methods. +3. Add class instances and attributes to log using the `track_variable` method. +4. Start logging data using the `update` method. +5. PLEASE call the `close` method before exiting the program to ensure all data is written to the log file. + +""" + +import csv +import logging +import os +from collections import deque +from datetime import datetime +from enum import Enum +from logging.handlers import RotatingFileHandler +from typing import Any, Callable, Optional, Union + +__all__ = ["LOGGER", "LOG_LEVEL", "Logger"] + + +class LogLevel(Enum): + DEBUG = logging.DEBUG + INFO = logging.INFO + WARNING = logging.WARNING + ERROR = logging.ERROR + CRITICAL = logging.CRITICAL + + +class Logger(logging.Logger): + _instance = None + + def __new__(cls, *args, **kwargs): + if cls._instance is None: + cls._instance = super().__new__(cls) + return cls._instance + + def __init__( + self, + log_path: str = "./", + log_format: str = "[%(asctime)s] %(levelname)s: %(message)s", + file_level: LogLevel = LogLevel.DEBUG, + stream_level: LogLevel = LogLevel.INFO, + file_max_bytes: int = 0, + file_backup_count: int = 5, + file_name: Union[str, None] = None, + buffer_size: int = 1000, + ) -> None: + if not hasattr(self, "_initialized"): + super().__init__(__name__) + self._log_path = log_path + self._log_format = log_format + self._file_level = file_level + self._stream_level = stream_level + self._file_max_bytes = file_max_bytes + self._file_backup_count = file_backup_count + self._user_file_name = file_name + + self._file_path: str = "" + self._csv_path: str = "" + self._file: Optional[Any] = None + self._writer = None + self._is_logging = False + self._header_written = False + + self._tracked_vars: dict[int, Callable[[], Any]] = {} + self._var_names: dict[int, str] = {} + self._buffer: deque[list[str]] = deque(maxlen=buffer_size) + self._buffer_size: int = buffer_size + + self._setup_logging() + self._initialized: bool = True + else: + self.set_file_name(file_name) + self.set_file_level(file_level) + self.set_stream_level(stream_level) + self.set_format(log_format) + self._file_max_bytes = file_max_bytes + self._file_backup_count = file_backup_count + self.set_buffer_size(buffer_size) + + self._log_path = log_path + + def _setup_logging(self) -> None: + self.setLevel(level=self._file_level.value) + self._std_formatter = logging.Formatter(self._log_format) + + self._stream_handler = logging.StreamHandler() + self._stream_handler.setLevel(level=self._stream_level.value) + self._stream_handler.setFormatter(fmt=self._std_formatter) + self.addHandler(hdlr=self._stream_handler) + + def _setup_file_handler(self) -> None: + if self._file_path == "": + self._generate_file_paths() + + self._file_handler = RotatingFileHandler( + filename=self._file_path, + mode="a", + maxBytes=self._file_max_bytes, + backupCount=self._file_backup_count, + ) + self._file_handler.setLevel(level=self._file_level.value) + self._file_handler.setFormatter(fmt=self._std_formatter) + self.addHandler(hdlr=self._file_handler) + + def _ensure_file_handler(self): + if not hasattr(self, "_file_handler"): + self._setup_file_handler() + + def track_variable(self, var_func: Callable[[], Any], name: str): + var_id = id(var_func) + self._tracked_vars[var_id] = var_func + self._var_names[var_id] = name + + def untrack_variable(self, var_func: Callable[[], Any]): + var_id = id(var_func) + self._tracked_vars.pop(var_id, None) + self._var_names.pop(var_id, None) + + def __repr__(self) -> str: + return f"Logger(file_path={self._file_path})" + + def set_file_name(self, file_name: Union[str, None]) -> None: + self._user_file_name = file_name + self._file_path = "" + self._csv_path = "" + + def set_file_level(self, level: LogLevel) -> None: + self._file_level = level + if hasattr(self, "_file_handler"): + self._file_handler.setLevel(level=level.value) + + def set_stream_level(self, level: LogLevel) -> None: + self._stream_level = level + self._stream_handler.setLevel(level=level.value) + + def set_format(self, log_format: str) -> None: + self._log_format = log_format + self._std_formatter = logging.Formatter(log_format) + if hasattr(self, "_file_handler"): + self._file_handler.setFormatter(fmt=self._std_formatter) + self._stream_handler.setFormatter(fmt=self._std_formatter) + + def set_buffer_size(self, buffer_size: int) -> None: + self._buffer_size = buffer_size + self._buffer = deque(self._buffer, maxlen=buffer_size) + + def update(self) -> None: + if not self._tracked_vars: + return + + data = [] + for _var_id, get_value in self._tracked_vars.items(): + value = get_value() + data.append(str(value)) + + self._buffer.append(data) + + if len(self._buffer) >= self._buffer_size: + self.flush_buffer() + + def flush_buffer(self): + if not self._buffer: + return + + self._ensure_file_handler() + + if self._file is None: + self._file = open(self._csv_path, "a", newline="") + self._writer = csv.writer(self._file) + + if not self._header_written: + self._write_header() + + self._writer.writerows(self._buffer) + self._buffer.clear() + self._file.flush() + + def _write_header(self) -> None: + header = list(self._var_names.values()) + + self._writer.writerow(header) # type: ignore[assignment] + self._header_written = True + + def _generate_file_paths(self) -> None: + now = datetime.now() + timestamp = now.strftime("%Y%m%d_%H%M%S") + script_name = os.path.basename(__file__).split(".")[0] + + base_name = self._user_file_name if self._user_file_name else f"{script_name}_{timestamp}" + + file_path = os.path.join(self._log_path, base_name) + self._file_path = file_path + ".log" + self._csv_path = file_path + ".csv" + + def __enter__(self) -> "Logger": + return self + + def __exit__(self, exc_type, exc_val, exc_tb) -> None: + self.flush_buffer() + self.close() + + def reset(self): + self._buffer.clear() + self._tracked_vars.clear() + self._var_names.clear() + self._header_written = False + if hasattr(self, "_file_handler"): + self._file_handler.close() + del self._file_handler + + def close(self) -> None: + if self._file: + self._file.close() + self._file = None + self._writer = None + + def debug(self, msg, *args, **kwargs): + self._ensure_file_handler() + super().debug(msg, *args, **kwargs) + + def info(self, msg, *args, **kwargs): + self._ensure_file_handler() + super().info(msg, *args, **kwargs) + + def warning(self, msg, *args, **kwargs): + self._ensure_file_handler() + super().warning(msg, *args, **kwargs) + + def error(self, msg, *args, **kwargs): + self._ensure_file_handler() + super().error(msg, *args, **kwargs) + + def critical(self, msg, *args, **kwargs): + self._ensure_file_handler() + super().critical(msg, *args, **kwargs) + + def log(self, level, msg, *args, **kwargs): + self._ensure_file_handler() + super().log(level, msg, *args, **kwargs) + + @property + def file_path(self) -> str: + if self._file_path == "": + self._generate_file_paths() + return self._file_path + + @property + def buffer_size(self) -> int: + return self._buffer_size + + @property + def file_level(self) -> LogLevel: + return self._file_level + + @property + def stream_level(self) -> LogLevel: + return self._stream_level + + @property + def file_max_bytes(self) -> int: + return self._file_max_bytes + + @property + def file_backup_count(self) -> int: + return self._file_backup_count + + +# Initialize a global logger instance to be used throughout the library +LOGGER = Logger() +LOG_LEVEL = dict(enumerate(LogLevel.__members__.values())) diff --git a/onshape_api/utils.py b/onshape_api/utils.py deleted file mode 100644 index 74e0724..0000000 --- a/onshape_api/utils.py +++ /dev/null @@ -1,64 +0,0 @@ -""" -utils -===== - -Handy functions for API key sample app -""" - -import logging -from logging.config import dictConfig - -__all__ = ["log"] - - -def log(msg, level=0): - """ - Logs a message to the console, with optional level paramater - - Args: - - msg (str): message to send to console - - level (int): log level; 0 for info, 1 for error (default = 0) - """ - - red = "\033[91m" - endc = "\033[0m" - - # configure the logging module - cfg = { - "version": 1, - "disable_existing_loggers": False, - "formatters": { - "stdout": { - "format": "[%(levelname)s]: %(asctime)s - %(message)s", - "datefmt": "%x %X", - }, - "stderr": { - "format": red + "[%(levelname)s]: %(asctime)s - %(message)s" + endc, - "datefmt": "%x %X", - }, - }, - "handlers": { - "stdout": { - "class": "logging.StreamHandler", - "level": "DEBUG", - "formatter": "stdout", - }, - "stderr": { - "class": "logging.StreamHandler", - "level": "ERROR", - "formatter": "stderr", - }, - }, - "loggers": { - "info": {"handlers": ["stdout"], "level": "INFO", "propagate": True}, - "error": {"handlers": ["stderr"], "level": "ERROR", "propagate": False}, - }, - } - - dictConfig(cfg) - - lg = "info" if level == 0 else "error" - lvl = 20 if level == 0 else 40 - - logger = logging.getLogger(lg) - logger.log(lvl, msg) diff --git a/pyproject.toml b/pyproject.toml index 1e41ba8..113a4f3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,6 +12,7 @@ packages = [ [tool.poetry.dependencies] python = ">=3.8,<4.0" +python-dotenv = "^1.0.1" [tool.poetry.group.dev.dependencies] pytest = "^7.2.0"