PK! qergal/__init__.py__version__ = '0.1.0' PK!7YNNergal/profile.py""" ergal Profile module. """ import os import json import hashlib import sqlite3 from warnings import warn from . import utils import xmltodict as xtd import requests from requests.exceptions import ConnectionError class Profile: """ Manages API profiles. """ def __init__(self, name, base=None, test=False): """ Initialize Profiler class. Profile handles the creation and storage of API profiles. These objects are created and stored in a local SQLite database called 'ergal.db'. Arguments: name -- the name of the profile Keyword Arguments: base -- the API's base URL test -- tells the util to create a test database that will be deleted following tests """ 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) 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 "Profile for {name} fetched from {id}.".format( name=self.name, id=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 "Profile for {name} created at {id}.".format( name=self.name, id=self.id) def call(self, name, **kwargs): """ Call an endpoint. Arguments: name -- the name of an endpoint """ endpoint = self.endpoints[name] url = self.base + endpoint['path'] 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) try: data = json.loads(response.text) except: data = xtd.parse(response.text) finally: 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 "Fields for {name} updated at {id}".format( name=self.name, id=self.id) def set_auth(self, method, **kwargs): """ Set authentication details. Arguments: method -- a supported auth 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 "Authentication details for {name} set at {id}".format( name=self.name, id=self.id) def add_endpoint(self, name, path, method, **kwargs): """ Add an endpoint. Arguments: name -- a name describing the given endpoint path -- the given path to the API endpoint method -- the method assigned to the given endpoint """ endpoint = {'path': path, 'method': method} for key in ('params', 'data', 'headers', 'auth'): 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 "Endpoint {path_name} for {name} added at {id}.".format( path_name=name, name=self.name, id=self.id) def del_endpoint(self, name): """ Delete an endpoint. Arguments: 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 "Endpoint {path} for {name} deleted from {id}.".format( path=name, name=self.name, id=self.id) PK!0<<ergal/utils.py""" ergal utilities. """ import os import json import sqlite3 import hashlib import xmltodict as xtd import requests 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. Arguments: name -- a str name of a database Keyword Arguments: test -- a bool defining whether or not to run for a test config Returns: db -- a database connection cursor -- a database cursor """ 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 PK!HlUTergal-0.1.6.dist-info/WHEEL HM K-*ϳR03rOK-J,/R(O-)T0343 /, (-JLR()*M IL*4KM̫PK!Hb&έhwergal-0.1.6.dist-info/METADATAWmoF_1E>$vgľIu'] Ւ\/R Q]R܇*H33 I/ :eS:I+QS:>8JN鶒JR4DSG EZ6 MoReSE+Jh/W?]%*ۻqS(xOD)+ Y /kQ:ΧG; Ixz~4~(ysj$_[SZQת)R4e1OY?hCώR0~>K Z\J '{/P})]nU둞 xS/%tWIW__\&gFZ*Оz+uM^YZA'} 2I5.MH544#BXT )DK6*@I76`W-6NHmU.8 ʡf19{gI XgMt}'xRqαJKG+ vPPUY^7|T^7O&u"7<X :9ϝC=ͨC.Y'It9ouUqZV׆<8$CV؂6 QO>jj?y EYsa@ueiƖK.sÓvNj:Z+ښVt:F{OIխ~2MOf?߼k-rY]P9qNf@ucMT6c`ED69:bVb`Uf v#rP| +Apr F)Z?݊A?>ҿMj-lj0U+|Bxb?惙Rk@wS>j}Yq؈({^Y<AeSe.$Pa~ LtGP1<a[L3`M(t1|ll?2URK0,gi8|C\7c0#4ɳzOi р7f{BlN :% /%U*?D8B,1Ce"|JAab )p; :8lZ'9;fZ|Fi5HlQ`Z1_^нL+sd.#'2-ZAO.1]3E5xg @11SM08?00 j8DLFq|*-,,J2|,Fj {i?s:LX*\HؐD/O?fBhdc`ˁ%[ǜzkN -{.--J-,4k]PIǞS.,.&yY`h7cHon]MWVQwƁ؉W"\%ѯ.#!{-9R {ά;=F+lXz3FWv8-[hY(.sC}".V6x,k5 zL1(;oBՊ0g.tVXu -yl).~O`Eo4 Tʚ/\.\ܺb,2r2H<&$"2u\w6,R+LcZkG>3NP$8:6K6-36XiR`)ڢoh˸)򱼠E;f3.!X^Q8ur(iQ:ķ[FK*p'נ5FBd_5o9 C1PK!H)N$ergal-0.1.6.dist-info/RECORDu;r@޳,˿H""d!ȟ z$cw\R. ~vY׏nxE(e/)M;|\T!p>I:yiT5 (6u A$NMi^?^ؔ.wb عԋSK-|0|C22P "}**i3Y3d O<5F.t_zp90f,vἵ%oʛCKPK! qergal/__init__.pyPK!7YNNEergal/profile.pyPK!0<<ergal/utils.pyPK!HlUT)ergal-0.1.6.dist-info/WHEELPK!Hb&έhwergal-0.1.6.dist-info/METADATAPK!H)N$[%ergal-0.1.6.dist-info/RECORDPK&