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

More code comments and type-setting

Browse source
This commit is contained in:
Elia El Lazkani 2019-11-04 17:43:31 +01:00
parent 3f42b5bb57
commit 28b68e9a74
6 changed files with 138 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

29
shortenit/config.py
View file

@ -2,21 +2,44 @@ import yaml
class Config:
"""
Configuration importer.
"""
def __init__(self, config_path: str):
"""
Initialize the configuration importer.
:param config_path: The path of the configuration file.
"""
self.config_path = config_path
self.config = None
def get_config(self):
def get_config(self) -> str:
"""
Get the configuration saved in the configuration file.
:returns: The configuration saved in the configuration file.
"""
if not self.config:
_config = self.load_config()
self.config = _config
return self.config
def load_config(self):
def load_config(self) -> str:
"""
Read the configuration saved in the configuration file.
:returns: The configuration saved in the configruation file.
"""
with open(self.config_path, 'rt') as f:
config = yaml.load(f)
if self.validate_config(config):
return config
def validate_config(self, config):
def validate_config(self, config: str) -> bool:
"""
Validate the configuration.
:returns: Configuration validation status.
"""
return True

34
shortenit/data.py
View file

@ -6,9 +6,19 @@ from cloudant.document import Document
class Data:
"""
Data object.
"""
def __init__(self, data_db: object,
identifier: str = None,
data: str = None):
data: str = None) -> None:
"""
Initialize the Data object.
:param data_db: The Data database object.
:param identifier: A uniquely generated ID identifying the data object.
:param data: The data to save.
"""
self.logger = logging.getLogger(self.__class__.__name__)
self.data_db = data_db
self.identifier = identifier
@ -18,11 +28,19 @@ class Data:
self.data_found = None
self.populate()
def generate_identifier(self):
def generate_identifier(self) -> None:
"""
Method to generate and save a new unique ID as the Data object identifier.
"""
hash_object = sha256(self.data.encode('utf-8'))
self.identifier = hash_object.hexdigest()
def populate(self, pointer: str = None):
def populate(self, pointer: str = None) -> None:
"""
Method to populate the Data object fields with proper data and save it in the database.
:param pointer: The unique ID of the pointer object to save with the data.
"""
if self.identifier:
self.logger.debug("The identifier is set, retrieving data...")
self.get_data()
@ -37,7 +55,10 @@ class Data:
"creating...")
self.set_data(pointer)
def get_data(self):
def get_data(self) -> None:
"""
Method to retrieve the Data ojbect from the database.
"""
with Document(self.data_db, self.identifier) as data:
try:
self.data = data['value']
@ -47,7 +68,10 @@ class Data:
except KeyError:
self.data_found = False
def set_data(self, pointer):
def set_data(self, pointer: str) -> None:
"""
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

19
shortenit/db.py
View file

@ -6,7 +6,15 @@ from cloudant.client import CouchDB
from shortenit.exceptions import DBConnectionFailed
class DB:
"""
Database object class
"""
def __init__(self, config: dict) -> None:
"""
Initialize the Database object.
:param config: The Database configuration.
"""
self.logger = logging.getLogger(self.__class__.__name__)
self.username = config['username']
self.password = config['password']
@ -14,7 +22,12 @@ class DB:
self.client = None
self.session = None
def initialize_shortenit(self):
def initialize_shortenit(self) -> None:
"""
Method to initialize the database for shortenit.
This will check if all the needed tables already exist in the database.
Otherwise, it will create the database tables.
"""
try:
self.data_db = self.client['data']
except KeyError:
@ -37,6 +50,9 @@ class DB:
def __enter__(self) -> CouchDB:
"""
Method used when entering the database context.
:returns: The CouchDB object.
"""
try:
self.client = CouchDB(self.username, self.password,
@ -54,5 +70,6 @@ class DB:
def __exit__(self, *args) -> None:
"""
Method used when exiting the database context.
"""
self.client.disconnect()

3
shortenit/exceptions.py
View file

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

30
shortenit/pointer.py
View file

@ -1,14 +1,27 @@
from __future__ import annotations
import time
import logging
from cloudant.document import Document
CHARS = '0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ'
class Pointer:
"""
Pointer object.
"""
def __init__(self, pointers_db: object,
identifier: str = None) -> None:
"""
Initialize the Pointer object.
:param pointers_db: The Pointer database object.
:param identifier: A uniquely generated ID identifying the pointer object.
"""
self.logger = logging.getLogger(self.__class__.__name__)
self.pointers_db = pointers_db
self.identifier = identifier
@ -16,7 +29,14 @@ class Pointer:
self.ttl = None
self.timestamp = time.time()
def generate_pointer(self, data_hash: str, ttl: time.time):
def generate_pointer(self, data_hash: str, ttl: time.time) -> Pointer:
"""
Generates a pointer object and saves it into the database.
:param data_hash: A uniquely generated ID identifying the data object.
:param ttl: The "Time to Live" of the pointer.
:returns: The Pointer object.
"""
self.logger.debug("identifier is %s", self.identifier)
with Document(self.pointers_db, self.identifier) as pointer:
pointer['value'] = data_hash
@ -26,7 +46,13 @@ class Pointer:
self.ttl = ttl
return self
def get_pointer(self, identifier: str):
def get_pointer(self, identifier: str) -> Pointer:
"""
Retrieve a pointer object from the database.
:param identifier: A uniquely generated ID identifying the Pointer object.
:returns: The Pointer object requested.
"""
with Document(self.pointers_db, identifier) as pointer:
try:
self.identifier = pointer['_id']

39
shortenit/shortener.py
View file

@ -5,7 +5,16 @@ from cloudant.document import Document
class Shortener:
def __init__(self, pointer_db, configuration: dict):
"""
Shortener object
"""
def __init__(self, pointer_db, configuration: dict) -> None:
"""
Initialize the Shortener object.
:param pointer_db: The Pointer Database object.
:param configuration: The shortenit configuration.
"""
self.logger = logging.getLogger(self.__class__.__name__)
self.pointer_db = pointer_db
self.uuid = None
@ -15,7 +24,10 @@ class Shortener:
self.configuration = configuration
self.init()
def init(self):
def init(self) -> None:
"""
Initialize the shortener from the configuration.
"""
length = self.configuration.get("id_length", 32)
if length >= 32 or length <= 0:
self.length = 32
@ -26,13 +38,23 @@ class Shortener:
self.upper_case = self.configuration.get(
"id_upper_case", False)
def generate_short_uuid(self):
def generate_short_uuid(self) -> str:
"""
Generate a short UUID in Hex format.
:returns: A short UUID in Hex format.
"""
short_uuid = uuid.uuid1().hex
if self.upper_case:
return short_uuid.upper()[0:self.length]
return short_uuid.lower()[0:self.length]
def check_uuid(self, short_uuid):
def check_uuid(self, short_uuid) -> bool:
"""
Check a short UUID against the database.
:returns: Whether the UUID exists in the database or not.
"""
with Document(self.pointer_db, 'pointer') as pointer:
self.logger.debug("Pointer: %s", pointer)
try:
@ -44,7 +66,13 @@ class Shortener:
return False
return True
def get_id(self):
def get_id(self) -> str:
"""
Method to get a UUID.
This method will generate a UUID and checks if it already exists in the database.
:returns: A UUID.
"""
short_uuid = self.generate_short_uuid()
if self.check_duplicate:
counter = 0
@ -57,4 +85,3 @@ class Shortener:
counter += 1
self.logger.debug("Returning ID: '%s'", short_uuid)
return short_uuid