Method | Purpose | Return Type |
find_token(self, token) | Given a token, return a full Token (OAuthToken) object from storage. Used before making any CRM API call. | Token(OAuthToken) object |
save_token(self, token) | Called right after Zoho returns a new access/refresh token. Your implementation must persist it. | None |
delete_token(self, id) | Delete a specific token using its unique ID. | None |
get_tokens(self) | Return all stored tokens. | A list of Token(OAuthToken) objects |
delete_tokens() | Delete all stored tokens. Useful during cleanup or logout. | None |
find_token_by_id(id) | Retrieve a token by its unique identifier. | Token(OAuthToken) object |
def __init__(self): """ Initializes the SQLite database and sets up the oauthtoken table if needed. """ self.db_name = 'zohooauth.db' if not self.check_table_exists(): connection = sqlite3.connect(self.db_name) cursor = connection.cursor() cursor.execute("CREATE TABLE oauthtoken (id varchar(10) NOT NULL,user_name varchar(255), client_id " "varchar(255), client_secret varchar(255), refresh_token varchar(255), access_token " "varchar(255), grant_token varchar(255), expiry_time varchar(20), redirect_url varchar(" "255), api_domain varchar(255), primary key (id))") cursor.close() |
def save_token(self, token): if not isinstance(token, OAuthToken): return cursor = None connection = None try: connection = sqlite3.connect(self.db_name) oauth_token = token query = "update oauthtoken set " if oauth_token.get_user_signature() is not None: name = oauth_token.get_user_signature().get_name() if name is not None and len(name) > 0: query = query + self.set_token(oauth_token) + " where user_name='" + name + "'" elif oauth_token.get_access_token() is not None and len(oauth_token.get_access_token()) > 0 and \ self.are_all_objects_null([oauth_token.get_client_id(), oauth_token.get_client_secret()]): query = query + self.set_token( oauth_token) + " where access_token='" + oauth_token.get_access_token() + "'" elif ((oauth_token.get_refresh_token() is not None and len(oauth_token.get_refresh_token()) > 0) or (oauth_token.get_grant_token() is not None and len( oauth_token.get_grant_token()) > 0)) and oauth_token.get_client_id() is not None \ and oauth_token.get_client_secret() is not None: if oauth_token.get_grant_token() is not None and len(oauth_token.get_grant_token()) > 0: query = query + self.set_token( oauth_token) + " where grant_token='" + oauth_token.get_grant_token() + "'" elif oauth_token.get_refresh_token() is not None and len(oauth_token.get_refresh_token()) > 0: query = query + self.set_token( oauth_token) + " where refresh_token='" + oauth_token.get_refresh_token() + "'" query = query + " limit 1" try: cursor = connection.cursor() cursor.execute(query) if cursor.rowcount <= 0: if oauth_token.get_id() is not None or oauth_token.get_user_signature() is not None: if oauth_token.get_refresh_token() is None and oauth_token.get_grant_token() is None \ and oauth_token.get_access_token() is None: raise SDKException(Constants.TOKEN_STORE, Constants.GET_TOKEN_DB_ERROR1) if oauth_token.get_id() is None: newId = str(self.generate_id()) oauth_token.set_id(newId) query = "insert into oauthtoken (id,user_name,client_id,client_secret,refresh_token,access_token," \ "grant_token,expiry_time,redirect_url,api_domain) values (?,?,?,?,?,?,?,?,?,?);" val = (token.get_id(), token.get_user_signature().get_name() if token.get_user_signature() is not None else None, token.get_client_id(), token.get_client_secret(), token.get_refresh_token(), token.get_access_token(), token.get_grant_token(), token.get_expires_in(), token.get_redirect_url(), token.get_api_domain()) cursor.execute(query, val) except Error as e: raise e finally: connection.commit() cursor.close() if cursor is not None else None connection.close() if connection is not None else None except Exception as ex: raise SDKException(Constants.TOKEN_STORE, Constants.SAVE_TOKEN_DB_ERROR, cause=ex) |
def find_token(self, token): cursor = None connection = None try: connection = sqlite3.connect(self.db_name) if isinstance(token, OAuthToken): oauth_token = token query = "select * from oauthtoken" if oauth_token.get_user_signature() is not None: name = oauth_token.get_user_signature().get_name() if name is not None and len(name) > 0: query = query + " where user_name='" + name + "'" elif oauth_token.get_access_token() is not None and self.are_all_objects_null( [oauth_token.get_client_id(), oauth_token.get_client_secret()]): query = query + " where access_token='" + oauth_token.get_access_token() + "'" elif oauth_token.get_refresh_token() is not None or oauth_token.get_grant_token() is not None and \ oauth_token.get_client_id() is not None and oauth_token.get_client_secret() is not None: if oauth_token.get_grant_token() is not None and len(oauth_token.get_grant_token()) > 0: query = query + " where grant_token='" + oauth_token.get_grant_token() + "'" elif oauth_token.get_refresh_token() is not None and len(oauth_token.get_refresh_token()) > 0: query = query + " where refresh_token='" + oauth_token.get_refresh_token() + "'" query = query + " limit 1" cursor = connection.cursor() cursor.execute(query) result = cursor.fetchone() if result is None: return None self.set_merge_data(oauth_token, result) except Exception as ex: raise SDKException(Constants.TOKEN_STORE, Constants.GET_TOKEN_DB_ERROR1, cause=ex) finally: cursor.close() if cursor is not None else None connection.close() if connection is not None else None return token |
def delete_token(self, id): cursor = None try: connection = sqlite3.connect(self.db_name) try: cursor = connection.cursor() query = "delete from oauthtoken where id= " + id + ";" cursor.execute(query) connection.commit() except Error as ex: raise ex finally: cursor.close() if cursor is not None else None connection.close() if connection is not None else None except Error as ex: raise SDKException(code=Constants.TOKEN_STORE, message=Constants.DELETE_TOKEN_DB_ERROR, cause=ex) |
def delete_tokens(self): cursor = None try: connection = sqlite3.connect(self.db_name) try: cursor = connection.cursor() query = "delete from oauthtoken;" cursor.execute(query) self.connection.commit() except Error as ex: raise ex finally: cursor.close() if cursor is not None else None connection.close() if connection is not None else None except Error as ex: raise SDKException(code=Constants.TOKEN_STORE, message=Constants.DELETE_TOKENS_DB_ERROR, cause=ex) |
def get_tokens(self): cursor = None try: connection = sqlite3.connect(self.db_name) tokens = [] try: cursor = connection.cursor() query = "select * from oauthtoken;" cursor.execute(query) results = cursor.fetchall() for result in results: oauth_token = object.__new__(OAuthToken) self.set_oauth_token(oauth_token) self.set_merge_data(oauth_token, result) tokens.append(oauth_token) return tokens except Error as ex: raise ex finally: cursor.close() if cursor is not None else None connection.close() if connection is not None else None except Error as ex: raise SDKException(code=Constants.TOKEN_STORE, message=Constants.GET_TOKENS_DB_ERROR, cause=ex) |
def find_token_by_id(self, id): cursor = None try: connection = sqlite3.connect(self.db_name) try: query = "select * from oauthtoken where id='" + id + "'" oauth_token = object.__new__(OAuthToken) self.set_oauth_token(oauth_token) cursor = connection.cursor() cursor.execute(query) results = cursor.fetchall() if results is None or len(results) <= 0: raise SDKException(Constants.TOKEN_STORE, Constants.GET_TOKEN_BY_ID_DB_ERROR) for result in results: self.set_merge_data(oauth_token, result) return oauth_token except Error as ex: raise ex finally: cursor.close() if cursor is not None else None connection.close() if connection is not None else None except Error as ex: raise SDKException(code=Constants.TOKEN_STORE, message=Constants.GET_TOKEN_BY_ID_DB_ERROR, cause=ex) |
def init(self, client_id, code, location, redirect_url): environment = DataCenter.get(location) client_secret = "17565609051856218813123b9a98de52c301722b7d" logger = Logger.get_instance(level=Logger.Levels.INFO, file_path="./logs.txt") store = CustomStoreSQLite() token = OAuthToken(client_id=client_id, client_secret=client_secret, grant_token=code, redirect_url=redirect_url) Initializer.initialize(environment=environment, token=token, logger=logger, store=store) |
Writer is a powerful online word processor, designed for collaborative work.