added simple user control to the template

Author: jsonzilla

.env_example

@@ -3,4 +3,5 @@ PROJECT_VERSION="0.0.1"
 PROJECT_DESCRIPTION="<>"
 MONGO_URL="<>"
 DEFAULT_DATABASE="<>"
-
+ENABLE_ADMIN=True
+API_TOKEN="<>"

.vscode/launch.json

@@ -2,12 +2,14 @@
   "version": "0.2.0",
   "configurations": [
     {
-      "name": "Run server",
+      "name": "Python: Module",
       "type": "python",
       "request": "launch",
-      "program": "./main.py",
-      "console": "integratedTerminal",
-      "justMyCode": true
+      "module": "uvicorn",
+      "args": [
+        "app.main:app",
+        "--reload"
+      ]
     },
     {
       "name": "PyTest",
@@ -17,15 +19,15 @@
       //USE IF NEEDED "python": "${command:python.interpreterPath}",
       "module": "pytest",
       "args": [
-          "-sv"
+        "-sv"
       ],
       "cwd": "${workspaceRoot}",
       "env": {},
       "envFile": "${workspaceRoot}/.env",
       "debugOptions": [
-          "WaitOnAbnormalExit",
-          "WaitOnNormalExit",
-          "RedirectOutput"
+        "WaitOnAbnormalExit",
+        "WaitOnNormalExit",
+        "RedirectOutput"
       ]
     }
   ]

.vscode/settings.json

@@ -1,6 +1,27 @@
 {
+  "python.linting.flake8Enabled": true,
+  "python.linting.flake8Args": [
+    "--max-line-length=120",
+    "--ignore=E501",
+    "--exclude=*.env,*.env_example"
+  ],
+  "editor.formatOnSave": true,
+  "python.testing.pytestArgs": [
+    "tests"
+  ],
+  "python.testing.unittestEnabled": false,
+  "python.testing.pytestEnabled": true,
+  "python.formatting.provider": "none",
+  "files.exclude": {
+    "**/__pycache__": true,
+    "**/*.pyc": true,
+    "**/.git": true,
+  },
+  "search.exclude": {
+    "**/*.pyc": true,
+    "**/.git": true,
+  },
   "[python]": {
     "editor.defaultFormatter": "ms-python.python"
-  },
-  "python.formatting.provider": "none"
+  }
 }

app/actions/verify_identify_action.py

@@ -0,0 +1,84 @@
+from fastapi import HTTPException, Request
+from app.models.data_identification_model import DataIdentificationModel
+from app.repositories import UserDataRepository, DefaultDataRepository
+
+
+_user_repo = UserDataRepository()
+_default_repo = DefaultDataRepository()
+
+
+class VerifyIdentityAction(object):
+    def extract_identifier(self, identifier_header) -> DataIdentificationModel:
+        """
+        Extracts the identifier from the header and returns it as a DataIdentificationModel
+        :param identifier_header: The identifier header
+        :return: The identifier as a DataIdentificationModel
+        """
+        identifier = identifier_header.split(' ')
+        if len(identifier) != 3:
+            raise HTTPException(status_code=422, detail=f"Invalid identifier: {identifier_header}")
+        return DataIdentificationModel(application=identifier[0], release=identifier[1], name=identifier[2])
+
+    def extract_client(self, client_header) -> str:
+        """
+        Extracts the client from the header and returns it as a string
+        :param client_header: The client header
+        :return: The client as a string
+        """
+        if client_header is None:
+            raise HTTPException(status_code=422, detail="Missing client identifier")
+        return client_header
+
+    async def find_client_data_by_header(self, request: Request, db):
+        """
+        Finds the data for the client and identifier in the request
+        :param request: The request
+        :param db: The database
+        :return: The data for the client and identifier in the request
+        """
+        client = self.extract_client(request.headers.get('client'))
+        identifier = self.extract_identifier(request.headers.get('identifier'))
+        db_filter = identifier.create_filter()
+        db_filter['client'] = client
+        return await _user_repo.get_one(db_filter, db)
+
+    async def find_default_data_by_header(self, request: Request, db):
+        """
+        Finds the default data for the identifier in the request
+        :param request: The request
+        :param db: The database
+        :return: The default data for the identifier in the request
+        """
+        identifier = self.extract_identifier(request.headers.get('identifier'))
+        db_filter = identifier.create_filter()
+        return await _default_repo.get_one(db_filter, db)
+
+    async def exist_default_data_by_header(self, request: Request, db):
+        """
+        Checks if the default data for the identifier in the request exists
+        :param request: The request
+        :param db: The database
+        :return: True if the default data exists, False otherwise
+        """
+        identifier = self.extract_identifier(request.headers.get('identifier'))
+        db_filter = identifier.create_filter()
+        return await _default_repo.exists(db_filter, db)
+
+    async def exist_default_data(self, identifier: DataIdentificationModel, db):
+        """
+        Checks if the default data for the identifier in the request exists
+        :param identifier: The identifier
+        :return: True if the default data exists, False otherwise
+        """
+        db_filter = identifier.create_filter()
+        return await _default_repo.exists(db_filter, db)
+
+    async def exist_user_data(self, identifier: DataIdentificationModel, db):
+        """
+        Checks if the user data for the identifier in the request exists
+        :param identifier: The identifier
+        :return: True if the default data exists, False otherwise
+        """
+        db_filter = identifier.create_filter()
+        res = await _user_repo.exists(db_filter, db)
+        return res

app/core/__init__.py

@@ -1 +1,2 @@
 from .config import settings
+from .hasher import Hasher

app/core/config.py

@@ -8,6 +8,8 @@ class Settings(BaseSettings):
 
     MONGO_URL: str = Field(env="MONGO_URL", default="mongodb://localhost:27017/", description="The url of the MongoDB")
     DEFAULT_DATABASE: str = Field(env="DEFAULT_DATABASE", default="your_database", description="Default database name")
+    ENABLE_ADMIN: bool = Field(env="ENABLE_ADMIN", default=True)
+    API_TOKEN: str = Field(env="API_TOKEN", default="your_token", description="The token for the API")
 
     class Config:
         validate_assignment = True

app/core/hasher.py

@@ -0,0 +1,13 @@
+from passlib.context import CryptContext
+
+_pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+
+class Hasher:
+    @staticmethod
+    def verify_password(plain_password, hashed_password):
+        return _pwd_context.verify(plain_password, hashed_password)
+
+    @staticmethod
+    def get_password_hash(password):
+        return _pwd_context.hash(password)

app/core/security.py

@@ -0,0 +1,58 @@
+import secrets
+from fastapi import Depends, HTTPException, status, Header
+from fastapi.security import HTTPBasic, HTTPBasicCredentials
+from app.core import Hasher, settings
+from app.repositories import UserRepository
+from app.storages.database_storage import get_db
+
+_security = HTTPBasic()
+
+
+async def get_token_header(x_token: str = Header(...)):
+    """
+    Get the token data from the header
+    :param x_token: str
+    :return: None
+    """
+    if x_token != settings.API_TOKEN:
+        raise HTTPException(status_code=401, detail="Missing Authorization Header")
+
+
+async def get_token_app_header(x_token: str = Header(...), identifier: str = Header(...), client: str = Header(...)):
+    """
+    Get the token and application identifier data from the header
+    :param x_token: str
+    :param identifier: str
+    :param client: str
+    :return: None
+    """
+    if x_token != settings.API_TOKEN:
+        raise HTTPException(status_code=401, detail="Missing Authorization Header")
+    if identifier is None:
+        raise HTTPException(status_code=401, detail="Missing Identifier")
+    if client is None:
+        raise HTTPException(status_code=401, detail="Missing Client")
+
+
+async def validate_auth(credentials: HTTPBasicCredentials = Depends(_security), db=Depends(get_db)):
+    """
+    Validate the user credentials
+    :param credentials: HTTPBasicCredentials
+    :param db: Database
+    :return: str username
+    """
+    repo = UserRepository()
+    user = await repo.get_one({'username': credentials.username}, db)
+
+    if user:
+        correct_username = secrets.compare_digest(credentials.username, user['username'])
+        correct_password = Hasher.verify_password(credentials.password, user['password'])
+
+        if (correct_username and correct_password):
+            return credentials.username
+
+    raise HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Incorrect username or password",
+        headers={"WWW-Authenticate": "Basic"},
+    )

app/main.py

@@ -1,8 +1,10 @@
-from fastapi import FastAPI
+from fastapi import FastAPI, Depends
 from fastapi.middleware.gzip import GZipMiddleware
 from app.routes import base_route
+from app.routes.v1.admin import user_route
 from app.routes.v1.person_route import router
 from app.core.config import settings
+from app.core.security import get_token_header
 from app.storages.database_storage import close_db, connect_db
 
 
@@ -12,8 +14,14 @@ def get_application():
         title=settings.PROJECT_NAME,
         version=settings.PROJECT_VERSION,
         description=settings.PROJECT_DESCRIPTION,
-        debug=False
+        debug=False,
+        swagger_ui_parameters={
+            "defaultModelsExpandDepth": -1,
+            "tagsSorter": "alpha",
+        },
     )
+    if (settings.ENABLE_ADMIN):
+        _app.include_router(user_route.router, dependencies=[Depends(get_token_header)])
     _app.include_router(base_route.router)
     _app.include_router(router)
     return _app

app/models/__init__.py

@@ -0,0 +1,2 @@
+from .user_model import UserModel, UpdateUserModel, ShowUserModel, CreateUserModel
+from .person_model import PersonModel, PersonUpdateModel, PersonFilterModel

app/models/examples/user_example.py

@@ -0,0 +1,21 @@
+example = {
+    "username": "johndoe",
+    "password": "$2y$10$Ve5lrAVaVQydb42YuUawOeBg/6JCrR4qXxGMw2BfhWkAniLiJZzuy",
+    "email": "johndoe@example.com",
+    "last_update_datetime": "2020-04-23 10:20:30.40000",
+}
+
+another_example = {
+    "username": "johndoejr",
+    "password": "$2y$10$Ve5lrAVaVQydb42YuUawOeBg/6JCrR4qXxGMw2BfhWkAniLiJZzuy",
+    "email": "johndoejr@example.com",
+    "last_update_datetime": "2020-04-13 10:20:30.40000",
+}
+
+# not use a hashed password
+
+basic_request_example = {
+    "username": "johndoe",
+    "password": "@BeatifullPass4g3",
+    "email": "johndoe@example.com",
+}

app/models/user_model.py

@@ -0,0 +1,71 @@
+from typing import Optional
+from datetime import datetime
+from bson import ObjectId
+from pydantic import BaseModel, Field, EmailStr, validator
+from app.codecs import ObjectIdCodec
+from app.models.examples.user_example import example, basic_request_example
+
+
+class UserModel(BaseModel):
+    id: ObjectIdCodec = Field(default_factory=ObjectIdCodec, alias="_id")
+    username: str = Field(unique_items=None, min_length=3, max_length=50, description="The username of the user.")
+    password: str = Field(description="The password of the user.", min_length=10, max_length=63)
+    email: EmailStr = Field(unique_items=None, description="The email of the user.")
+    last_update_datetime: Optional[datetime] = Field(default_factory=datetime.utcnow, description="The date and time that this user was last updated.")
+
+    @validator('username')
+    def username_alphanumeric(cls, v):
+        assert v.isalnum(), 'must be alphanumeric'
+        return v
+
+    class Config:
+        allow_population_by_field_name = True
+        json_encoders = {ObjectId: str, datetime: str}
+        schema_extra = {"example": basic_request_example}
+        extra = 'forbid'
+        copy_on_model_validation = 'none'
+
+
+class CreateUserModel(BaseModel):
+    id: ObjectIdCodec = Field(default_factory=ObjectIdCodec, alias="_id")
+    username: str = Field(unique_items=None, min_length=3, max_length=50, description="The username of the user.")
+    password: str = Field(description="The password of the user.", min_length=10, max_length=63)
+    email: EmailStr = Field(unique_items=None, description="The email of the user.")
+
+    @validator('username')
+    def username_alphanumeric(cls, v):
+        assert v.isalnum(), 'must be alphanumeric'
+        return v
+
+    class Config:
+        allow_population_by_field_name = True
+        json_encoders = {ObjectId: str, datetime: str}
+        schema_extra = {"example": basic_request_example}
+        extra = 'forbid'
+        copy_on_model_validation = 'none'
+
+
+class ShowUserModel(BaseModel):
+    id: ObjectIdCodec = Field(default_factory=ObjectIdCodec, alias="_id")
+    username: str = Field(unique_items=None, min_length=3, max_length=50)
+    email: EmailStr = Field(unique_items=None, description="The email of the user.")
+    last_update_datetime: datetime = Field(default_factory=datetime.utcnow,
+                                           description="The date and time that this user was last updated.")
+
+    class Config:
+        allow_population_by_field_name = True
+        json_encoders = {ObjectId: str, datetime: str}
+        schema_extra = {"example": example}
+        extra = 'forbid'
+        copy_on_model_validation = 'none'
+
+
+class UpdateUserModel(BaseModel):
+    password: Optional[str] = Field(description="The password of the user.", min_length=10, max_length=63)
+    last_update_datetime: Optional[datetime] = Field(description="The date and time that this user was last updated.")
+
+    class Config:
+        json_encoders = {ObjectId: str, datetime: str}
+        schema_extra = {"example": example}
+        extra = 'forbid'
+        copy_on_model_validation = 'none'

app/repositories/__init__.py

@@ -0,0 +1,2 @@
+from .person_repository import PersonRepository
+from .user_repository import UserRepository

app/repositories/user_repository.py

@@ -0,0 +1,6 @@
+from app.repositories.base_repository import BaseRepository
+
+
+class UserRepository(BaseRepository):
+    def __init__(self):
+        super().__init__('user')

app/routes/__init__.py

@@ -1 +1 @@
-from .routable.basic_routable import BasicRoutable
+from .basic_router import BasicRouter