diff --git a/shortenit/common.py b/shortenit/common.py index 9d520d9..48b37c7 100644 --- a/shortenit/common.py +++ b/shortenit/common.py @@ -1,6 +1,5 @@ -import os import logging - +import os # Setup logging logger = logging.getLogger(__name__) diff --git a/shortenit/config.py b/shortenit/config.py index 819a23e..cadd350 100644 --- a/shortenit/config.py +++ b/shortenit/config.py @@ -5,6 +5,7 @@ class Config: """ Configuration importer. """ + def __init__(self, config_path: str): """ Initialize the configuration importer. @@ -31,7 +32,7 @@ class Config: :returns: The configuration saved in the configruation file. """ - with open(self.config_path, 'rt') as f: + with open(self.config_path, "rt") as f: config = yaml.safe_load(f) if self.validate_config(config): return config diff --git a/shortenit/data.py b/shortenit/data.py index fb61614..84f33ed 100644 --- a/shortenit/data.py +++ b/shortenit/data.py @@ -1,8 +1,8 @@ -import typing -import time import logging - +import time +import typing from hashlib import sha256 + from cloudant.document import Document @@ -10,9 +10,10 @@ class Data: """ Data object. """ - def __init__(self, data_db: object, - identifier: str = None, - data: str = None) -> typing.NoReturn: + + def __init__( + self, data_db: object, identifier: str = None, data: str = None + ) -> typing.NoReturn: """ Initialize the Data object. @@ -33,7 +34,7 @@ class Data: """ Method to generate and save a new unique ID as the Data object identifier. """ - hash_object = sha256(self.data.encode('utf-8')) + hash_object = sha256(self.data.encode("utf-8")) self.identifier = hash_object.hexdigest() def populate(self, pointer: str = None) -> typing.NoReturn: @@ -48,12 +49,12 @@ class Data: elif self.data: self.logger.debug("The data is set, generating an identifier...") self.generate_identifier() - self.logger.debug("Attempting to get the data with " - "the identifier generated...") + self.logger.debug( + "Attempting to get the data with " "the identifier generated..." + ) self.get_data() if not self.data_found: - self.logger.debug("The data generated is not found, " - "creating...") + self.logger.debug("The data generated is not found, " "creating...") self.set_data(pointer) def get_data(self) -> typing.NoReturn: @@ -62,9 +63,9 @@ class Data: """ with Document(self.data_db, self.identifier) as data: try: - self.data = data['value'] - self.timestamp = data['timestamp'] - self.pointers = data['pointers'] + self.data = data["value"] + self.timestamp = data["timestamp"] + self.pointers = data["pointers"] self.data_found = True except KeyError: self.data_found = False @@ -74,9 +75,9 @@ class Data: Method to save Data object to the database. """ with Document(self.data_db, self.identifier) as data: - data['value'] = self.data - data['timestamp'] = self.timestamp + data["value"] = self.data + data["timestamp"] = self.timestamp try: - data['pointers'].append(pointer) + data["pointers"].append(pointer) except KeyError: - data['pointers'] = [pointer] + data["pointers"] = [pointer] diff --git a/shortenit/db.py b/shortenit/db.py index 640fc7f..983515b 100644 --- a/shortenit/db.py +++ b/shortenit/db.py @@ -1,15 +1,17 @@ -import typing import logging -import requests +import typing +import requests from cloudant.client import CouchDB from shortenit.exceptions import DBConnectionFailed + class DB: """ Database object class """ + def __init__(self, config: dict) -> typing.NoReturn: """ Initialize the Database object. @@ -17,9 +19,9 @@ class DB: :param config: The Database configuration. """ self.logger = logging.getLogger(self.__class__.__name__) - self.username = config['username'] - self.password = config['password'] - self.url = config['url'] + self.username = config["username"] + self.password = config["password"] + self.url = config["url"] self.client = None self.session = None @@ -30,24 +32,20 @@ class DB: Otherwise, it will create the database tables. """ try: - self.data_db = self.client['data'] + self.data_db = self.client["data"] except KeyError: - self.logger.warn( - "The 'data' database was not found, creating...") - self.data_db = self.client.create_database('data') + self.logger.warn("The 'data' database was not found, creating...") + self.data_db = self.client.create_database("data") if self.data_db.exists(): - self.logger.info( - "The 'data' database was successfully created.") + self.logger.info("The 'data' database was successfully created.") try: - self.pointers_db = self.client['pointers'] + self.pointers_db = self.client["pointers"] except KeyError: - self.logger.warn( - "The 'pointers' database was not found, creating...") - self.pointers_db = self.client.create_database('pointers') + self.logger.warn("The 'pointers' database was not found, creating...") + self.pointers_db = self.client.create_database("pointers") if self.pointers_db.exists(): - self.logger.info( - "The 'pointers' database was successfully created.") + self.logger.info("The 'pointers' database was successfully created.") def __enter__(self) -> CouchDB: """ @@ -56,8 +54,9 @@ class DB: :returns: The CouchDB object. """ try: - self.client = CouchDB(self.username, self.password, - url=self.url, connect=True) + self.client = CouchDB( + self.username, self.password, url=self.url, connect=True + ) except requests.exceptions.ConnectionError as e: self.logger.fatal("Failed to connect to database, is it on?") self.logger.fatal("%s", e) diff --git a/shortenit/exceptions.py b/shortenit/exceptions.py index ef7ca28..a48cfae 100644 --- a/shortenit/exceptions.py +++ b/shortenit/exceptions.py @@ -1,7 +1,9 @@ import sys + class DBConnectionFailed(Exception): """ DBConnectionFailed class exception. """ + pass diff --git a/shortenit/logger.py b/shortenit/logger.py index 3639557..a65c48c 100644 --- a/shortenit/logger.py +++ b/shortenit/logger.py @@ -1,13 +1,17 @@ +import logging.config import os import typing + import yaml -import logging.config + from .common import check_file -def setup_logging(default_path: str = None, - default_level: int = logging.ERROR, - env_key: str = 'LOG_CFG') -> typing.NoReturn: +def setup_logging( + default_path: str = None, + default_level: int = logging.ERROR, + env_key: str = "LOG_CFG", +) -> typing.NoReturn: """ Method that sets the logging system up. @@ -30,10 +34,12 @@ def setup_logging(default_path: str = None, path = None if path: - with open(path, mode='r') as f: + with open(path, mode="r") as f: config = yaml.safe_load(f.read()) logging.config.dictConfig(config) else: - _format = '%(asctime)s - %(levelname)s - %(filename)s:' \ - '%(name)s.%(funcName)s:%(lineno)d - %(message)s' + _format = ( + "%(asctime)s - %(levelname)s - %(filename)s:" + "%(name)s.%(funcName)s:%(lineno)d - %(message)s" + ) logging.basicConfig(level=default_level, format=_format) diff --git a/shortenit/main.py b/shortenit/main.py index 28d7a63..8e38257 100644 --- a/shortenit/main.py +++ b/shortenit/main.py @@ -1,24 +1,23 @@ #!/usr/bin/env python3 -import sys -import typing import argparse +import asyncio import logging import pathlib -import asyncio - +import sys import time +import typing -from shortenit.data import Data -from shortenit.pointer import Pointer from shortenit.config import Config -from shortenit.shortener import Shortener +from shortenit.data import Data from shortenit.db import DB -from shortenit.logger import setup_logging -from shortenit.web import Web, SiteHandler from shortenit.exceptions import DBConnectionFailed +from shortenit.logger import setup_logging +from shortenit.pointer import Pointer +from shortenit.shortener import Shortener +from shortenit.web import SiteHandler, Web PROJECT_ROOT = pathlib.Path(__file__).parent.parent -CONFIGURATION = f'{PROJECT_ROOT}/config/config.yaml' +CONFIGURATION = f"{PROJECT_ROOT}/config/config.yaml" # Setup logging logger = logging.getLogger(__name__) @@ -34,8 +33,8 @@ def main() -> typing.NoReturn: verbosity_level = verbosity(args.verbose) setup_logging(args.logger, verbosity_level) config = Config(CONFIGURATION).get_config() - db_config = config.get('CouchDB', None) - server_config = config.get('Server', None) + db_config = config.get("CouchDB", None) + server_config = config.get("Server", None) if db_config: try: with DB(db_config) as db: @@ -43,22 +42,19 @@ def main() -> typing.NoReturn: handler = SiteHandler(config, db, shorten_url, lenghten_url) web = Web(handler, debug=debug) - web.host = server_config.get('host', None) - web.port = server_config.get('port', None) + web.host = server_config.get("host", None) + web.port = server_config.get("port", None) web.start_up() except DBConnectionFailed as e: sys.exit(1) sys.exit(0) -def shorten_url(configuration: dict, database: DB, - data: str, ttl): - shortener = Shortener(database.pointers_db, - configuration.get('Shortener', None)) +def shorten_url(configuration: dict, database: DB, data: str, ttl): + shortener = Shortener(database.pointers_db, configuration.get("Shortener", None)) identifier = shortener.get_id() if identifier: - _data = Data(database.data_db, - data=data) + _data = Data(database.data_db, data=data) _data.populate() pointer = Pointer(database.pointers_db, identifier) pointer.generate_pointer(_data.identifier, ttl) @@ -82,15 +78,15 @@ def argument_parse() -> argparse.ArgumentParser: :returns: The argument parser. """ parser = argparse.ArgumentParser( - description="Generates rundeck resources " - "file from different API sources.") + description="Generates rundeck resources " "file from different API sources." + ) parser.add_argument( - '-v', '--verbose', action='count', default=0, - help='Verbosity level to use.') + "-v", "--verbose", action="count", default=0, help="Verbosity level to use." + ) parser.add_argument( - '-l', '--logger', type=str, - help='The logger YAML configuration file.') + "-l", "--logger", type=str, help="The logger YAML configuration file." + ) return parser @@ -111,5 +107,5 @@ def verbosity(verbose: int): return logging.DEBUG -if __name__ == '__main__': +if __name__ == "__main__": main() diff --git a/shortenit/pointer.py b/shortenit/pointer.py index cecaee4..f1641cc 100644 --- a/shortenit/pointer.py +++ b/shortenit/pointer.py @@ -1,8 +1,8 @@ from __future__ import annotations +import logging import time import typing -import logging from cloudant.document import Document @@ -11,8 +11,8 @@ class Pointer: """ Pointer object. """ - def __init__(self, pointers_db: object, - identifier: str = None) -> typing.NoReturn: + + def __init__(self, pointers_db: object, identifier: str = None) -> typing.NoReturn: """ Initialize the Pointer object. @@ -36,9 +36,9 @@ class Pointer: """ self.logger.debug("identifier is %s", self.identifier) with Document(self.pointers_db, self.identifier) as pointer: - pointer['value'] = data_hash - pointer['ttl'] = ttl - pointer['timestamp'] = self.timestamp + pointer["value"] = data_hash + pointer["ttl"] = ttl + pointer["timestamp"] = self.timestamp self.data_hash = data_hash self.ttl = ttl return self @@ -52,10 +52,10 @@ class Pointer: """ with Document(self.pointers_db, identifier) as pointer: try: - self.identifier = pointer['_id'] - self.data_hash = pointer['value'] - self.ttl = pointer['ttl'] - self.timestamp = pointer['timestamp'] + self.identifier = pointer["_id"] + self.data_hash = pointer["value"] + self.ttl = pointer["ttl"] + self.timestamp = pointer["timestamp"] return self except KeyError: pass diff --git a/shortenit/shortener.py b/shortenit/shortener.py index 6e4f1b5..9879e78 100644 --- a/shortenit/shortener.py +++ b/shortenit/shortener.py @@ -1,6 +1,6 @@ -import uuid -import typing import logging +import typing +import uuid from cloudant.document import Document @@ -9,6 +9,7 @@ class Shortener: """ Shortener object """ + def __init__(self, pointer_db, configuration: dict) -> typing.NoReturn: """ Initialize the Shortener object. @@ -34,10 +35,8 @@ class Shortener: self.length = 32 else: self.length = length - self.check_duplicate = self.configuration.get( - "check_duplicate_id", False) - self.upper_case = self.configuration.get( - "id_upper_case", False) + self.check_duplicate = self.configuration.get("check_duplicate_id", False) + self.upper_case = self.configuration.get("id_upper_case", False) def generate_short_uuid(self) -> str: """ @@ -47,8 +46,8 @@ class Shortener: """ short_uuid = uuid.uuid1().hex if self.upper_case: - return short_uuid.upper()[0:self.length] - return short_uuid.lower()[0:self.length] + return short_uuid.upper()[0 : self.length] + return short_uuid.lower()[0 : self.length] def check_uuid(self, short_uuid) -> bool: """ @@ -56,14 +55,14 @@ class Shortener: :returns: Whether the UUID exists in the database or not. """ - with Document(self.pointer_db, 'pointer') as pointer: + with Document(self.pointer_db, "pointer") as pointer: self.logger.debug("Pointer: %s", pointer) try: self.uuid = pointer[short_uuid] except KeyError: - self.logger.info("Generated short uuid '%s'" - "was not found in database", - short_uuid) + self.logger.info( + "Generated short uuid '%s'" "was not found in database", short_uuid + ) return False return True @@ -79,8 +78,10 @@ class Shortener: counter = 0 while self.check_uuid(short_uuid): if counter > 10: - self.logger.err("Cannot generate new unique ID," - "try to configure a longer ID length.") + self.logger.err( + "Cannot generate new unique ID," + "try to configure a longer ID length." + ) return None short_uuid = self.generate_short_uuid() counter += 1 diff --git a/shortenit/web.py b/shortenit/web.py index ba4c8ba..eaf066d 100644 --- a/shortenit/web.py +++ b/shortenit/web.py @@ -1,14 +1,9 @@ -import os import logging +import os +from pathlib import Path import trafaret - -from pathlib import Path -from flask import Flask -from flask import render_template -from flask import request -from flask import redirect -from flask import abort +from flask import Flask, abort, redirect, render_template, request class Web: @@ -29,12 +24,13 @@ class Web: self.setup_routes() def setup_routes(self): - self.app.add_url_rule('/', '/', self.handler.index, - methods=['GET']) - self.app.add_url_rule('/shortenit', '/shortenit', self.handler.shortenit, - methods=['POST']) - self.app.add_url_rule('/', '/identifier', self.handler.short_redirect, - methods=['GET']) + self.app.add_url_rule("/", "/", self.handler.index, methods=["GET"]) + self.app.add_url_rule( + "/shortenit", "/shortenit", self.handler.shortenit, methods=["POST"] + ) + self.app.add_url_rule( + "/", "/identifier", self.handler.short_redirect, methods=["GET"] + ) class SiteHandler: @@ -44,10 +40,9 @@ class SiteHandler: self.database = database self.shorten_url = shorten_url self.lenghten_url = lenghten_url - self.shortenit_load_format = trafaret.Dict({ - trafaret.Key('url'): trafaret.URL, - trafaret.Key('timestamp'): trafaret.Int - }) + self.shortenit_load_format = trafaret.Dict( + {trafaret.Key("url"): trafaret.URL, trafaret.Key("timestamp"): trafaret.Int} + ) def shortenit(self): data = request.get_json() @@ -60,8 +55,8 @@ class SiteHandler: abort(400) try: short_url = self.shorten_url( - self.configuration, self.database, - data['url'], data['timestamp']) + self.configuration, self.database, data["url"], data["timestamp"] + ) except KeyError as e: self.logger.error(e) abort(400) @@ -77,4 +72,4 @@ class SiteHandler: return redirect(url) def index(self): - return render_template('index.html') + return render_template("index.html")