Moving to clipboard for copying to clipboard

.gitignore

@@ -1,5 +1,3 @@
 .eggs/
 *.egg-info
 __pycache__/
-.mypy*/
-.vscode/

LICENSE

@@ -1,25 +0,0 @@
-BSD 2-Clause License
-
-Copyright (c) 2019, Elia El Lazkani
-All rights reserved.
-
-Redistribution and use in source and binary forms, with or without
-modification, are permitted provided that the following conditions are met:
-
-* Redistributions of source code must retain the above copyright notice, this
-  list of conditions and the following disclaimer.
-
-* Redistributions in binary form must reproduce the above copyright notice,
-  this list of conditions and the following disclaimer in the documentation
-  and/or other materials provided with the distribution.
-
-THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
-AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
-IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
-DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
-FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
-DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
-SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
-CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
-OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
-OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

README.md

@@ -1,22 +0,0 @@
-# Shortenit
-
-Shortenit is a tool to shorten urls.
-
-**NOTE**: This is a very early draft project. Contributions are welcome.
-
-## Running
-
-To run `shortenit`, first we need to have a running database. Shortenit uses [CouchDB](https://couchdb.apache.org/) as a database. CouchDB can run as a docker container.
-
-```text
-$ docker run -p 5984:5984 -d couchdb 
-```
-
-At this point, visit the local instance link [http://localhost:5984](http://localhost:5984/) and create the user credentials configured in [config/config.yaml](config/config.yaml).
-
-Once the database is up and running and the credentials have been created, shortenit can be ran.
-
-```text
-$ pip install -e .
-$ shortenit
-```

README.rst

@@ -0,0 +1,6 @@
+shortenit
+=========
+
+shortenit is a tool to shorten urls.
+
+NOTE: This is a very early draft project. Contributions are welcome.

TODO

@@ -1,10 +0,0 @@
-This list is incomplete, these are a few of the things that need to be done.
-
-TODO:
-* Documentation/Code comments
-* Proper type-setting across the board
-* PEP8
-* Refactor code
-* Dockerize
-* favicon.ico is being injected into the database (need fix)
-* Fix YAML warning on startup.

config/config.yaml

@@ -5,10 +5,3 @@ CouchDB:
   username: root
   password: root
   url: http://localhost:5984
-Shortener:
-  # *CAUTION*: Enabling this check if the ID already exists before returning it.
-  # Even though this guarantees that the ID doesn't exist, this might inflict
-  # some performance hit.
-  check_duplicate_id: False
-  id_length: 32
-  id_upper_case: False

requirements/development.txt

@@ -1,4 +1,2 @@
 -r requirements.txt
 setuptools-git
-setuptools-git-version
-mypy

setup.py

@@ -4,7 +4,7 @@ from setuptools import setup
 #<link rel=stylesheet type=text/css href="{{ url('static', filename='css/custom.css') }}">
 here = Path(__file__).absolute().parent
 
-with open(str(here.joinpath('README.md')), encoding='utf-8') as f:
+with open(str(here.joinpath('README.rst')), encoding='utf-8') as f:
     long_description = f.read()
 
 with open(str(here.joinpath('requirements/requirements.txt')),
@@ -16,7 +16,7 @@ description = 'Shortenit will shorten that url for you'
 setup(
     name='shortenit',
     exclude_package_data={'': ['.gitignore', 'requirements/', 'setup.py']},
-    version_format='{gitsha}',
+    version_format='{tag}_{gitsha}',
     setup_requires=['setuptools-git', 'setuptools-git-version'],
     license='BSD',
     author='Elia El Lazkani',
@@ -40,7 +40,6 @@ setup(
         'Programming Language :: Python',
         'Programming Language :: Python :: 3.6',
         'Programming Language :: Python :: 3.7',
-        'Programming Language :: Python :: 3.10',
         'Environment :: Console',
         'Intended Audience :: Information Technology',
         'Intended Audience :: System Administrators'

shortenit/config.py

@@ -2,44 +2,21 @@ 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) -> str:
+    def get_config(self):
-        """
-        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) -> str:
+    def load_config(self):
-        """
-        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.safe_load(f)
+            config = yaml.load(f)
             if self.validate_config(config):
                 return config
 
-    def validate_config(self, config: str) -> bool:
+    def validate_config(self, config):
-        """
-        Validate the configuration.
-
-        :returns: Configuration validation status.
-        """
         return True

shortenit/counter.py

@@ -0,0 +1,29 @@
+import logging
+
+from cloudant.document import Document
+
+
+class Counter:
+    def __init__(self, counter_db):
+        self.logger = logging.getLogger(self.__class__.__name__)
+        self.counter_db = counter_db
+        self.counter = None
+
+    def get_counter(self) -> int:
+        with Document(self.counter_db, 'counter') as counter:
+            self.logger.debug("Counter: %s", counter)
+            try:
+                self.counter = counter['value']
+            except KeyError:
+                self.logger.warn(
+                    "Counter was not initialized, initializing...")
+                counter['value'] = 0
+            try:
+                counter['value'] += 1
+            except Exception as e:
+                self.logger.err(e)
+            # Need to check if the value exists or not as to not jump values
+            # which it currently does but it's not a big issue for right now
+            self.counter = counter['value']
+
+        return self.counter

shortenit/data.py

@@ -1,4 +1,3 @@
-import typing
 import time
 import logging
 
@@ -7,19 +6,9 @@ from cloudant.document import Document
 
 
 class Data:
-    """
-    Data object.
-    """
     def __init__(self, data_db: object,
                  identifier: str = None,
-                 data: str = None) -> typing.NoReturn:
+                 data: str = 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
@@ -29,19 +18,11 @@ class Data:
         self.data_found = None
         self.populate()
 
-    def generate_identifier(self) -> typing.NoReturn:
+    def generate_identifier(self):
-        """
-        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) -> typing.NoReturn:
+    def populate(self, pointer: str = 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()
@@ -56,10 +37,7 @@ class Data:
                                   "creating...")
                 self.set_data(pointer)
 
-    def get_data(self) -> typing.NoReturn:
+    def get_data(self):
-        """
-        Method to retrieve the Data ojbect from the database.
-        """
         with Document(self.data_db, self.identifier) as data:
             try:
                 self.data = data['value']
@@ -69,10 +47,7 @@ class Data:
             except KeyError:
                 self.data_found = False
 
-    def set_data(self, pointer: str) -> typing.NoReturn:
+    def set_data(self, pointer):
-        """
-        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

shortenit/db.py

@@ -1,21 +1,10 @@
-import typing
 import logging
-import requests
 
 from cloudant.client import CouchDB
 
-from shortenit.exceptions import DBConnectionFailed
 
 class DB:
-    """
+    def __init__(self, config: dict) -> None:
-    Database object class
-    """
-    def __init__(self, config: dict) -> typing.NoReturn:
-        """
-        Initialize the Database object.
-
-        :param config: The Database configuration.
-        """
         self.logger = logging.getLogger(self.__class__.__name__)
         self.username = config['username']
         self.password = config['password']
@@ -23,12 +12,17 @@ class DB:
         self.client = None
         self.session = None
 
-    def initialize_shortenit(self) -> typing.NoReturn:
+    def initialize_shortenit(self):
-        """
+        try:
-        Method to initialize the database for shortenit.
+            self.counter_db = self.client['counter']
-        This will check if all the needed tables already exist in the database.
+        except KeyError:
-        Otherwise, it will create the database tables.
+            self.logger.warn(
-        """
+                "The 'counter' database was not found, creating...")
+            self.counter_db = self.client.create_database('counter')
+            if self.counter_db.exists():
+                self.logger.info(
+                    "The 'counter' database was successfully created.")
+
         try:
             self.data_db = self.client['data']
         except KeyError:
@@ -51,26 +45,13 @@ 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,
-            self.client = CouchDB(self.username, self.password,
+                              url=self.url, connect=True)
-                                  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)
-            raise DBConnectionFailed
-        except requests.exceptions.HTTPError as e:
-            self.logger.fatal("Failed to authenticate to database.")
-            self.logger.fatal("%s", e)
-            raise DBConnectionFailed
         self.session = self.client.session()
         return self
 
-    def __exit__(self, *args) -> typing.NoReturn:
+    def __exit__(self, *args) -> None:
         """
-        Method used when exiting the database context.
         """
         self.client.disconnect()

shortenit/exceptions.py

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

shortenit/logger.py

@@ -1,5 +1,4 @@
 import os
-import typing
 import yaml
 import logging.config
 from .common import check_file
@@ -7,7 +6,7 @@ from .common import check_file
 
 def setup_logging(default_path: str = None,
                   default_level: int = logging.ERROR,
-                  env_key: str = 'LOG_CFG') -> typing.NoReturn:
+                  env_key: str = 'LOG_CFG') -> None:
     """
     Method that sets the logging system up.
 

shortenit/main.py

@@ -1,6 +1,4 @@
 #!/usr/bin/env python3
-import sys
-import typing
 import argparse
 import logging
 import pathlib
@@ -8,14 +6,13 @@ import asyncio
 
 import time
 
-from shortenit.data import Data
+from .data import Data
-from shortenit.pointer import Pointer
+from .pointer import Pointer
-from shortenit.config import Config
+from .config import Config
-from shortenit.shortener import Shortener
+from .counter import Counter
-from shortenit.db import DB
+from .db import DB
-from shortenit.logger import setup_logging
+from .logger import setup_logging
-from shortenit.web import Web, SiteHandler
+from .web import Web, SiteHandler
-from shortenit.exceptions import DBConnectionFailed
 
 PROJECT_ROOT = pathlib.Path(__file__).parent.parent
 CONFIGURATION = f'{PROJECT_ROOT}/config/config.yaml'
@@ -24,7 +21,7 @@ CONFIGURATION = f'{PROJECT_ROOT}/config/config.yaml'
 logger = logging.getLogger(__name__)
 
 
-def main() -> typing.NoReturn:
+def main() -> None:
     """
     Main method
     """
@@ -37,34 +34,28 @@ def main() -> typing.NoReturn:
     db_config = config.get('CouchDB', None)
     server_config = config.get('Server', None)
     if db_config:
-        try:
+        with DB(db_config) as db:
-            with DB(db_config) as db:
+            db.initialize_shortenit()
-                db.initialize_shortenit()
 
-                handler = SiteHandler(config, db, shorten_url, lenghten_url)
+            handler = SiteHandler(db, shorten_url, lenghten_url)
-                web = Web(handler, debug=debug)
+            web = Web(handler, debug=debug)
-                web.host = server_config.get('host', None)
+            web.host = server_config.get('host', None)
-                web.port = server_config.get('port', None)
+            web.port = server_config.get('port', None)
-                web.start_up()
+            web.start_up()
-        except DBConnectionFailed as e:
-            sys.exit(1)
-    sys.exit(0)
 
 
-def shorten_url(configuration: dict, database: DB,
+def shorten_url(database: DB, data: str, ttl: time.time):
-                data: str, ttl):
+    counter = Counter(database.counter_db)
-    shortener = Shortener(database.pointers_db, 
+    data = Data(database.data_db,
-        configuration.get('Shortener', None))
+                data=data)
-    identifier = shortener.get_id()
+    data.populate()
-    if identifier:
+    pointer = Pointer(database.pointers_db, counter)
-        _data = Data(database.data_db,
+    pointer.generate_pointer(
-                    data=data)
+        data.identifier,
-        _data.populate()
+        ttl
-        pointer = Pointer(database.pointers_db, identifier)
+        )
-        pointer.generate_pointer(_data.identifier, ttl)
+    data.set_data(pointer.identifier)
-        _data.set_data(pointer.identifier)
+    return pointer.identifier
-        return pointer.identifier
-    return None
 
 
 def lenghten_url(database: DB, identifier: str):
@@ -109,7 +100,3 @@ def verbosity(verbose: int):
         return logging.INFO
     elif verbose > 2:
         return logging.DEBUG
-
-
-if __name__ == '__main__':
-    main()

shortenit/pointer.py

@@ -1,40 +1,29 @@
-from __future__ import annotations
-
 import time
-import typing
 import logging
 
+from .counter import Counter
 from cloudant.document import Document
 
+CHARS = '0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ'
+
 
 class Pointer:
-    """
-    Pointer object.
-    """
     def __init__(self, pointers_db: object,
-                 identifier: str = None) -> typing.NoReturn:
+                 counter: Counter = 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
+        self.counter = counter
+        self.identifier = None
         self.data_hash = None
         self.ttl = None
         self.timestamp = time.time()
 
-    def generate_pointer(self, data_hash: str, ttl: time.time) -> Pointer:
+    def generate_pointer(self, data_hash: str, ttl: time.time):
-        """
+        self.logger.debug("Generating new counter...")
-        Generates a pointer object and saves it into the database.
+        counter = self.counter.get_counter()
-
+        self.logger.debug("Encoding the counter into an ID")
-        :param data_hash: A uniquely generated ID identifying the data object.
+        self.identifier = Pointer.encode(counter)
-        :param ttl: The "Time to Live" of the pointer.
+        self.logger.debug("Encoded counter is %s", self.identifier)
-        :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
             pointer['ttl'] = ttl
@@ -43,13 +32,7 @@ class Pointer:
             self.ttl = ttl
         return self
 
-    def get_pointer(self, identifier: str) -> Pointer:
+    def get_pointer(self, identifier: str):
-        """
-        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']
@@ -60,3 +43,25 @@ class Pointer:
             except KeyError:
                 pass
         return None
+
+    @staticmethod
+    def encode(counter):
+        sign = '-' if counter < 0 else ''
+        counter = abs(counter)
+        result = ''
+        while counter > 0:
+            counter, remainder = divmod(counter, len(CHARS))
+            result = CHARS[remainder]+result
+
+        return sign+result
+
+    @staticmethod
+    def decode(counter):
+        return int(counter, len(CHARS))
+
+    @staticmethod
+    def padding(counter: str, count: int=6):
+        if len(counter) < count:
+            pad = '0' * (count - len(counter))
+            return f"{pad}{counter}"
+        return f"{counter}"

shortenit/shortener.py

@@ -1,88 +0,0 @@
-import uuid
-import typing
-import logging
-
-from cloudant.document import Document
-
-
-class Shortener:
-    """
-    Shortener object
-    """
-    def __init__(self, pointer_db, configuration: dict) -> typing.NoReturn:
-        """
-        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
-        self.length = 32
-        self.check_duplicate = False
-        self.upper_case = False
-        self.configuration = configuration
-        self.init()
-
-    def init(self) -> typing.NoReturn:
-        """
-        Initialize the shortener from the configuration.
-        """
-        length = self.configuration.get("id_length", 32)
-        if length >= 32 or length <= 0:
-            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)
-
-    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) -> 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:
-                self.uuid = pointer[short_uuid]
-            except KeyError:
-                self.logger.info("Generated short uuid '%s'"
-                                 "was not found in database",
-                                 short_uuid)
-                return False
-            return True
-
-    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
-            while self.check_uuid(short_uuid):
-                if counter > 10:
-                    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
-        self.logger.debug("Returning ID: '%s'", short_uuid)
-        return short_uuid

shortenit/web.py

@@ -38,9 +38,8 @@ class Web:
 
 
 class SiteHandler:
-    def __init__(self, configuration, database, shorten_url, lenghten_url):
+    def __init__(self, database, shorten_url, lenghten_url):
         self.logger = logging.getLogger(self.__class__.__name__)
-        self.configuration = configuration
         self.database = database
         self.shorten_url = shorten_url
         self.lenghten_url = lenghten_url
@@ -60,8 +59,7 @@ class SiteHandler:
             abort(400)
         try:
             short_url = self.shorten_url(
-                self.configuration, self.database,
+                self.database, data['url'], data['timestamp'])
-                data['url'], data['timestamp'])
         except KeyError as e:
             self.logger.error(e)
             abort(400)