elia/shortenit
1
0
Fork 0
Code Issues 4 Pull requests 1 Releases Packages Wiki Activity

chore(): Formats the code

Browse source 
* Formatting the code
* Fixing imports
This commit is contained in:
Elia el Lazkani 2024-11-30 11:07:02 +01:00
parent b2ca72a809
commit f10af22168
10 changed files with 119 additions and 119 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
shortenit/common.py
View file

@ -1,6 +1,5 @@
import os
import logging
import os
# Setup logging
logger = logging.getLogger(__name__)

3
shortenit/config.py
View file

@ -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

37
shortenit/data.py
View file

@ -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]

37
shortenit/db.py
View file

@ -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)

2
shortenit/exceptions.py
View file

@ -1,7 +1,9 @@
import sys
class DBConnectionFailed(Exception):
"""
DBConnectionFailed class exception.
"""
pass

18
shortenit/logger.py
View file

@ -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,
def setup_logging(
default_path: str = None,
default_level: int = logging.ERROR,
env_key: str = 'LOG_CFG') -> typing.NoReturn:
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)

50
shortenit/main.py
View file

@ -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()

20
shortenit/pointer.py
View file

@ -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

25
shortenit/shortener.py
View file

@ -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:
"""
@ -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

37
shortenit/web.py
View file

@ -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>', '/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>", "/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")