PK!R~ergal/__init__.py__version__ = '0.1.0' PK!WWergal/profile.py""" ergal.profile ~~~~~~~~~~~~~ This module implements the Profile interface, which enables the user to manage their API profiles. :copyright: (c) 2018 by Elliott Maguire """ import os import json import hashlib import sqlite3 from . import utils import requests import xmltodict as xtd class Profile: """ Enables API profile management. This class handles the creation/storage/management of API profiles in a local SQLite3 database called `ergal.db`, unless it is instantiated as a test instance, in which case the database is called `ergal_test.db`. :param name: a name for the API profile :param base: (optional) the base URL of the API :param test: (optional) dictates whether or not the database instance created should be a test instance. Example: >>> profile = Profile('HTTPBin', base='https://httpbin.com') >>> profile.add_endpoint('JSON', '/json', 'get') >>> profile.call('JSON') """ def __init__(self, name, base=None, test=False): self.name = name if type(name) is str else 'default' make_id = lambda n: ( hashlib.sha256(bytes(n, 'utf-8')) .hexdigest()[::2][::2]) self.id = make_id(name) if type(name) is str else 'default' self.base = base if type(base) is str else 'default' self.auth = {} self.endpoints = {} self.db, self.cursor = utils.get_db(test=test) try: self._get() except Exception as e: if str(e) == 'get: no matching record': self._create() else: raise Exception('get/create: unknown error occurred') def _get(self): """ Get an existing profile.. """ sql = """SELECT * FROM Profile WHERE id = ?""" self.cursor.execute(sql, (self.id,)) record = self.cursor.fetchone() if record: self.id = record[0] self.name = record[1] self.base = record[2] self.auth = json.loads(record[3]) if record[3] else {} self.endpoints = json.loads(record[4]) if record[4] else {} else: raise Exception('get: no matching record') return f"Profile for {self.name} fetched from {self.id}." def _create(self): """ Create a new profile. """ sql = "INSERT INTO Profile (id, name, base) VALUES (?, ?, ?)" with self.db: self.cursor.execute(sql, (self.id, self.name, self.base,)) return f"Profile for {self.name} created at {self.id}." def call(self, name, **kwargs): """ Call an endpoint. This method preps request items (url, headers, body), then makes a call to the given endpoint. The response is then parsed by `utils.parse` to produce an output. :param name: the name of the endpoint """ endpoint = self.endpoints[name] url = self.base + endpoint['path'] targets = endpoint['targets'] if 'targets' in endpoint else None if 'auth' in endpoint and endpoint['auth']: auth = endpoint['auth'] if auth['method'] == 'header': kwargs['headers'][auth['name']] = auth['key'] elif auth['method'] == 'params': kwargs['params'][auth['name']] = auth['key'] elif auth['method'] == 'basic': kwargs['auth'] = (auth['user'], auth['pass']) for k in kwargs: if k not in ('headers', 'params', 'data'): kwargs.pop(k) response = getattr(requests, endpoint['method'])(url, **kwargs) data = utils.parse(response, targets=targets) print(data) return data def update(self): """ Update the current profile's record. """ fields = vars(self) for field in fields.items(): sql = "UPDATE Profile SET ? = ? WHERE id = ?" with self.db: self.cursor.execute(sql, (field[0], field[1], self.id,)) return f"Fields for {self.name} updated at {self.id}" def set_auth(self, method, **kwargs): """ Set authentication details. :param method: a supported authentication method """ auth = {'method': method} for k, v in kwargs.items(): if k in ('key', 'name', 'username', 'password'): auth[k] = v self.auth = auth auth_str = json.dumps(self.auth) sql = "UPDATE Profile SET auth = ? WHERE id = ?" with self.db: self.cursor.execute(sql, (auth_str, self.id,)) return f"Authentication details for {self.name} set at {self.id}" def add_endpoint(self, name, path, method, **kwargs): """ Add an endpoint. :param name: a name for the endpoint :param path: the path, from the base URL, to the endpoint :param method: a supported HTTP method """ endpoint = {'path': path, 'method': method} for key in ('params', 'data', 'headers', 'auth', 'targets'): if key in kwargs: endpoint[key] = kwargs[key] else: continue self.endpoints[name] = endpoint endpoints_str = json.dumps(self.endpoints) sql = "UPDATE Profile SET endpoints = ? WHERE id = ?" with self.db: self.cursor.execute(sql, (endpoints_str, self.id,)) return f"Endpoint {name} for {self.name} added at {self.id}." def del_endpoint(self, name): """ Delete an endpoint. :param name: the name of an endpoint """ del self.endpoints[name] endpoints_str = json.dumps(self.endpoints) sql = "UPDATE Profile SET endpoints = ? WHERE id = ?" with self.db: self.cursor.execute(sql, (endpoints_str, self.id,)) return f"Endpoint {name} for {self.name} deleted from {self.id}." def add_target(self, endpoint, target): """ Add a data target. :param endpoint: the name of the endpoint :param target: the name of the target field """ targets = ( self.endpoints[endpoint]['targets'] if 'targets' in self.endpoints[endpoint] else []) targets.append(target) self.endpoints[endpoint]['targets'] = targets endpoints_str = json.dumps(self.endpoints) sql = "UPDATE Profile SET endpoints = ? WHERE id = ?" with self.db: self.cursor.execute(sql, (endpoints_str, self.id,)) return f"Target {target} for {endpoint} deleted from {self.id}." PK!I@ergal/utils.py""" ergal.utils ~~~~~~~~~~~ This module implements the utility methods used by the Profile interface. :copyright: (c) 2018 by Elliott Maguire """ import os import json import types import sqlite3 import hashlib import requests import xmltodict as xtd def get_db(test=False): """ Get/create a database connection. If a local ergal.db file exists, a connection is established and returned, otherwise a new database is created and the connection is returned. :param test: (optional) determines whether or not a test database should be created. """ file = 'ergal_test.db' if test else 'ergal.db' db = sqlite3.connect(file) cursor = db.cursor() db.execute(""" CREATE TABLE IF NOT EXISTS Profile ( id TEXT NOT NULL, name TEXT NOT NULL, base TEXT NOT NULL, auth TEXT, endpoints TEXT, PRIMARY KEY(id))""") return db, cursor def parse(response, targets=None): """ Parse response data. :param response: a requests.Response object :param targets: (optional) a list of data targets """ try: data = json.loads(response.text) except: data = xtd.parse(response.text) def search(d): for k in d: if k in targets: yield d elif type(d) is str: continue elif type(d[k]) in [dict, list]: for i in d[k]: if i in targets: yield d[k][i] for j in search(i): yield j result = search(data) if targets else data output = ( [i for i in result] if type(result) is types.GeneratorType else result) return output PK!HlUTergal-0.1.7.dist-info/WHEEL HM K-*ϳR03rOK-J,/R(O-)T0343 /, (-JLR()*M IL*4KM̫PK!Hϋergal-0.1.7.dist-info/METADATAXmoF_1E>$v+jľs4ɝ㸱[8%"7^rݥ{fEQT$gf^,~)ӜIz\ZQz~vu-pJw7Z$>I(ʴVhyOW*NޙZ.ZQBOx\\6nE+cOT+=}e/d-SEԹp>mO>X-nHwߦGɥΩP|cMiE]+єSUlVΟRn!AgG)\㣭UJwr6(TdT@ U'ΤX?JNϦ'www7T3l6a_΀5(kzВc)" E'R)g|9@[ǿo?^3 ,Qv$hL耄Ȏ^.(;j1TdC (X` < bS+ ">+X5Ob{ EP^f0p` [!<l!1T䡂[=rlX@Q3ᆅ7؇EbC(;sp8fU=a$ )c": ~31IٙaGcyvp; H" (;l3lLj_R3Qi&g,[ʦ& Eӗ݋h2W|@1 zӢ/r uYNH؅D/OQ3F!@tO@怭gcB_5mA=W|[JayĉfvU5.t c *:a,R۝1\wnE5nC>ōB Cjzhz.餍F`t \X e )Q19߱xWX0|㓚-c3Š/,*^c-߾7.v#w¢ G .Mx9L>6[:EqXl- MEom7AU&]]bssf8I) 0HgA6HSt,clɥVx--&G`7sWVP1.&n #:.W v)*&Њ M̩yWVM,-^̹zI_^Dp"'2eڌc &lc!ząxG"zӦB[$d Ne.e|]]vWk>%n߃l6# حZԉDʡ]FC:/M?KKpc@!5o[Ko]e4PK!H: 'ergal-0.1.7.dist-info/RECORDuɒC@{]!A0 .DPbT/n8+c16 ^X5t{J+w0Rn6( 3ah[vߚڶA9tZvnCCy=,kB?޾5>pS=mcޙ)OejUēc Ј A4)׬투UKWWUuCDHñbo`Ds!aブQӟ"z}w#)#RꞸFl[uMcx9G9}cm8F@Z[PK!R~ergal/__init__.pyPK!WWFergal/profile.pyPK!I@ergal/utils.pyPK!HlUT#ergal-0.1.7.dist-info/WHEELPK!Hϋ!$ergal-0.1.7.dist-info/METADATAPK!H: '>,ergal-0.1.7.dist-info/RECORDPK-