project basic init

Author: alisharify7

server/Flask-Backend/.env.sample

@@ -0,0 +1,44 @@
+APP_DEBUG=True
+APP_SECRET_KEY=supersecure
+ADMIN_LOGIN_TOKEN=random
+
+SMS_API_KEY=
+SMS_LINE_NUMBER=
+
+CAPTCHA_PUBLIC_KEY_V2=
+CAPTCHA_PRIVATE_KEY_V2=
+CAPTCHA_ENABLED_V2=True
+CAPTCHA_LOG_V2=True
+CAPTCHA_LANGUAGE_V2=en
+
+CAPTCHA_PUBLIC_KEY_V3=
+CAPTCHA_PRIVATE_KEY_V3=
+CAPTCHA_ENABLED_V3=True
+CAPTCHA_SCORE_V3=0.5
+CAPTCHA_LOG_V3=True
+
+CACHE_TYPE=NullCache
+
+REDIS_DEFAULT_URI=redis://localhost:6379/0
+REDIS_SESSION_URI=redis://localhost:6379/1
+REDIS_CELERY_BACKEND_URI=redis://localhost:6379/2
+REDIS_CELERY_BROKER_URI=redis://localhost:6379/3
+REDIS_CACHE_URI=redis://localhost:6379/4
+REDIS_LIMIT_URI=redis://localhost:6379/5
+
+DATABASE_USERNAME=
+DATABASE_PASSWORD=
+DATABASE_PORT=3306
+DATABASE_HOST=localhost
+DATABASE_NAME=mini
+DATABASE_TABLE_PREFIX_NAME=mini_
+
+MAIL_SERVER=
+MAIL_PORT=587
+MAIL_USERNAME=
+MAIL_PASSWORD=
+MAIL_USE_TLS=True
+MAIL_DEBUG=False
+MAIL_DEFAULT_SENDER=[email protected]
+
+SERVER_NAME=localhost:8000

server/Flask-Backend/app.py

@@ -0,0 +1,5 @@
+from core import app
+
+
+if __name__ == "__main__":
+    app.run()

server/Flask-Backend/config/__init__.py

@@ -0,0 +1 @@
+from .setting import Setting

server/Flask-Backend/config/setting.py

@@ -2,6 +2,7 @@
 import datetime
 from pathlib import Path
 
+
 import redis
 from dotenv import load_dotenv
 from core.utils import generate_random_string
@@ -21,6 +22,10 @@ class Setting:
 
     SECRET_KEY = os.environ.get("APP_SECRET_KEY", generate_random_string())
 
+    API_TITLE = "UrlShorter API"
+    API_VERSION = "v1"
+    OPENAPI_VERSION = "3.0.2"
+
     ADMIN_LOGIN_TOKEN = os.environ.get("ADMIN_LOGIN_TOKEN", "123654")
 
     APP_DEBUG_STATUS = os.environ.get("APP_DEBUG", "") == "True"
@@ -47,7 +52,7 @@ class Setting:
     DATABASE_USERNAME = os.environ.get("DATABASE_USERNAME", "")
     DATABASE_PASSWORD = os.environ.get("DATABASE_PASSWORD", "")
     DATABASE_TABLE_PREFIX_NAME = os.environ.get("DATABASE_TABLE_PREFIX_NAME", "")
-    SQLALCHEMY_DATABASE_URI = f"mysql+pymysql://{DATABASE_USERNAME}:{DATABASE_PASSWORD}@{DATABASE_HOST}:{DATABASE_PORT}/{DATABASE_NAME}"
+    SQLALCHEMY_DATABASE_URI = f"sqlite:///database.sqlite3"
     SQLALCHEMY_TRACK_MODIFICATIONS = False
 
     # Redis Config

server/Flask-Backend/core/__init__.py

@@ -0,0 +1,33 @@
+from flask import Flask
+from werkzeug.middleware.proxy_fix import ProxyFix
+from config import Setting
+
+from .extensions import api, db, ServerMigrate
+from .urls import urlpatterns
+
+
+
+def create_app(setting: Setting) -> Flask:
+    """
+        Factory Function For creating FlaskApp
+    """
+    app = Flask(__name__,)
+
+
+    app.config.from_object(setting)
+
+    db.init_app(app=app)  # db
+    ServerMigrate.init_app(db=extensions.db, app=app)  # migrate
+    api.init_app(app=app)
+
+    for each in urlpatterns:
+        api.register_blueprint(each["obj"], url_prefix=each["url_prefix"])
+
+    app.wsgi_app = ProxyFix(  # tell flask in behind a reverse proxy
+        app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_prefix=1
+    )
+    return app
+
+
+app = create_app(Setting)
+

server/Flask-Backend/core/extensions.py

@@ -0,0 +1,10 @@
+# flask extensions
+
+from flask_migrate import Migrate
+from flask_sqlalchemy import SQLAlchemy
+from flask_smorest import Api
+
+
+api = Api()
+db = SQLAlchemy()
+ServerMigrate = Migrate()

server/Flask-Backend/core/models.py

@@ -0,0 +1,65 @@
+import uuid
+import datetime
+from typing import Optional
+
+import sqlalchemy as sa
+import sqlalchemy.orm as so
+from flask import current_app
+
+from config import Setting
+from .extensions import db
+
+class BaseModel(db.Model):
+    """
+    Base model class for all models
+     ~~~~~~~~~~~~~~ abstract model ~~~~~~~~~~~~~~~
+
+    """
+
+    __abstract__ = True
+    # __table_args__ = {
+    #     # 'mysql_engine': 'InnoDB',
+    #     # 'mysql_charset': 'utf8',
+    #     # 'mysql_collate': 'utf8_persian_ci'
+    # }
+
+    id: so.Mapped[int] = so.mapped_column(sa.INTEGER, primary_key=True)
+    @staticmethod
+    def SetTableName(name):
+        """Use This Method For setting a table name"""
+        name = name.replace("-", "_").replace(" ", "")
+        return f"{Setting.DATABASE_TABLE_PREFIX_NAME}{name}".lower()
+
+    def set_public_key(self):
+        """ This Method Set a Unique PublicKey """
+        while True:
+            token = uuid.uuid4().hex
+            if self.query.filter_by(public_key=token).first():
+                continue
+            else:
+                self.public_key = token
+                break
+
+    def save(self, show_traceback: bool = True):
+        """
+         combination of two steps, add and commit session
+        """
+        try:
+            db.session.add(self)
+            db.session.commit()
+        except Exception as e:
+            db.session.rollback()
+            if show_traceback:
+                current_app.logger.exception(exc_info=e, msg=e)
+            return False
+        else:
+            return True
+
+    public_key: so.Mapped[str] = so.mapped_column(sa.String(36), nullable=False, unique=True)
+    created_time: so.Mapped[Optional[datetime.datetime]] = so.mapped_column(sa.DateTime,
+                                                                           default=datetime.datetime.now)
+    modified_time: so.Mapped[Optional[datetime.datetime]] = so.mapped_column(sa.DateTime,
+                                                                              onupdate=datetime.datetime.now,
+                                                                              default=datetime.datetime.now)
+
+

server/Flask-Backend/core/urls.py

@@ -0,0 +1,5 @@
+from shorter import blp as ShorterBluePrint
+
+urlpatterns = [
+    {"obj": ShorterBluePrint, "url_prefix": "/links/"}
+]

server/Flask-Backend/core/utils.py

@@ -1,22 +1,13 @@
-import uuid
-import datetime
-import string
 import random
-from urllib.parse import urlparse as url_parse
-from celery import Celery, Task
-from flask import Flask, current_app, request, session
-from werkzeug.utils import secure_filename as werkzeug_secure_filename
-
-SysRandom = random.SystemRandom()
+import string
 
 
-def generate_random_string(length: int = 6, punctuation: bool = True) -> str:
+SysRandom = random.SystemRandom()
+def generate_random_string(length: int = 3, punctuation: bool = False) -> str:
     """generate random strings
-
      params:
         length: int  = length of random string - default is 6
         punctuation: bool = punctuation in random string or not
-
      return:
         str: string: random string
      """
@@ -26,44 +17,3 @@ def generate_random_string(length: int = 6, punctuation: bool = True) -> str:
     random_string = SysRandom.choices(letters, k=length)
 
     return "".join(random_string)
-
-
-def get_next_page(fall_back_url: str = '') -> str:
-    """
-    use this method for validating next params in url
-    validate http url args next=some url
-    """
-    next_page = request.args.get("next", False)
-    if not next_page or url_parse(next_page).netloc != "":
-        next_page = fall_back_url
-
-    return next_page
-
-
-
-def make_file_name_secure(name: str, round:int=3):
-    """This function make sure a file name is secure
-    remove dangerous characters and add  uuid to first of file name
-    """
-    name = name.replace(" ", "")
-    name = werkzeug_secure_filename(name)
-    return f"{''.join([uuid.uuid4().hex for _ in range(round)])}-{datetime.datetime.utcnow().date()}-{name}"
-
-
-def celery_init_app(app: Flask) -> Celery:
-    class FlaskTask(Task):
-        """Every time a task is added to queue
-         __call__ ...
-        """
-
-        def __call__(self, *args: object, **kwargs: object) -> object:
-            with app.app_context():
-                return self.run(*args, **kwargs)
-
-    celery_app = Celery(app.name, task_cls=FlaskTask)
-    celery_app.config_from_object(app.config["CELERY"])
-    celery_app.Task = FlaskTask
-    celery_app.set_default()
-    app.extensions["celery"] = celery_app
-    return celery_app
-

server/Flask-Backend/shorter/models.py

@@ -0,0 +1,61 @@
+import sqlalchemy as sa
+import sqlalchemy.orm as so
+
+from core.extensions import db
+from core.models import BaseModel
+from core.utils import generate_random_string
+
+class AliasLink(BaseModel):
+    __tablename__ = BaseModel.SetTableName("alias-links")
+
+    alias_link: so.Mapped[str] = so.mapped_column(sa.String(2048), unique=True, nullable=False)
+    link_id: so.Mapped[int] = so.mapped_column(sa.INTEGER, sa.ForeignKey(BaseModel.SetTableName("links")+".id", ondelete="CASCADE"), nullable=False, unique=False)
+
+
+class Link(BaseModel):
+    __tablename__ = BaseModel.SetTableName("links")
+
+    url_address: so.Mapped[str] = so.mapped_column(sa.String(2048), unique=True, nullable=False)
+    alias = so.relationship("AliasLink", backref="url_address", lazy="joined")
+
+    @staticmethod
+    def generate_alias_url(link_id):
+        """generate unique alias url """
+        default_length = 3
+        counter = 0
+        total_round = 0
+
+        while True:
+            if total_round == 20:
+                return None
+
+            if counter == 5:
+                counter = 0
+                total_round += 1
+                default_length += 1
+
+            slug_token = generate_random_string(length=default_length)
+            query = db.select(AliasLink).filter_by(alias_link=slug_token)
+            result = db.session.execute(query).scalar_one_or_none()
+            if result: # duplicated alias link
+                counter += 1
+                continue
+            else:
+                obj = AliasLink(alias_link=slug_token, link_id=link_id)
+                obj.set_public_key()
+                obj.save()
+                return obj
+
+
+    @classmethod
+    def generate_alias_link(cls, url: str) -> AliasLink:
+        query = db.select(Link).filter_by(url_address=url).distinct()
+        result = db.session.execute(query).unique().scalar_one_or_none()
+        if result:
+            return result.alias
+
+        new_link = Link(url_address=url)
+        new_link.set_public_key()
+        new_link.save()
+        alias_obj = cls.generate_alias_url(new_link.id)
+        return alias_obj