API - add typing
This commit is contained in:
@@ -1,30 +1,44 @@
|
||||
import re
|
||||
from datetime import timedelta
|
||||
from functools import wraps
|
||||
from typing import Any, Callable, Optional, Tuple, Union
|
||||
|
||||
import humanize
|
||||
from fittrackee.responses import (
|
||||
ForbiddenErrorResponse,
|
||||
HttpResponse,
|
||||
InvalidPayloadErrorResponse,
|
||||
PayloadTooLargeErrorResponse,
|
||||
UnauthorizedErrorResponse,
|
||||
)
|
||||
from flask import current_app, request
|
||||
from flask import Request, current_app, request
|
||||
|
||||
from .models import User
|
||||
|
||||
|
||||
def is_admin(user_id):
|
||||
def is_admin(user_id: int) -> bool:
|
||||
"""
|
||||
Return if user has admin rights
|
||||
"""
|
||||
user = User.query.filter_by(id=user_id).first()
|
||||
return user.admin
|
||||
|
||||
|
||||
def is_valid_email(email):
|
||||
def is_valid_email(email: str) -> bool:
|
||||
"""
|
||||
Return if email format is valid
|
||||
"""
|
||||
mail_pattern = r"(^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$)"
|
||||
return re.match(mail_pattern, email) is not None
|
||||
|
||||
|
||||
def check_passwords(password, password_conf):
|
||||
def check_passwords(password: str, password_conf: str) -> str:
|
||||
"""
|
||||
Verify if password and password confirmation are the same and have
|
||||
more than 8 characters
|
||||
|
||||
If not, it returns not empty string
|
||||
"""
|
||||
ret = ''
|
||||
if password_conf != password:
|
||||
ret = 'Password and password confirmation don\'t match.\n'
|
||||
@@ -33,7 +47,14 @@ def check_passwords(password, password_conf):
|
||||
return ret
|
||||
|
||||
|
||||
def register_controls(username, email, password, password_conf):
|
||||
def register_controls(
|
||||
username: str, email: str, password: str, password_conf: str
|
||||
) -> str:
|
||||
"""
|
||||
Verify if user name, email and passwords are valid
|
||||
|
||||
If not, it returns not empty string
|
||||
"""
|
||||
ret = ''
|
||||
if not 2 < len(username) < 13:
|
||||
ret += 'Username: 3 to 12 characters required.\n'
|
||||
@@ -43,7 +64,12 @@ def register_controls(username, email, password, password_conf):
|
||||
return ret
|
||||
|
||||
|
||||
def verify_extension_and_size(file_type, req):
|
||||
def verify_extension_and_size(
|
||||
file_type: str, req: Request
|
||||
) -> Optional[HttpResponse]:
|
||||
"""
|
||||
Return error Response if file is invalid
|
||||
"""
|
||||
if 'file' not in req.files:
|
||||
return InvalidPayloadErrorResponse('No file part.', 'fail')
|
||||
|
||||
@@ -66,7 +92,7 @@ def verify_extension_and_size(file_type, req):
|
||||
|
||||
if not (
|
||||
file_extension
|
||||
and file_extension in current_app.config.get(allowed_extensions)
|
||||
and file_extension in current_app.config[allowed_extensions]
|
||||
):
|
||||
return InvalidPayloadErrorResponse(
|
||||
'File extension not allowed.', 'fail'
|
||||
@@ -81,7 +107,13 @@ def verify_extension_and_size(file_type, req):
|
||||
return None
|
||||
|
||||
|
||||
def verify_user(current_request, verify_admin):
|
||||
def verify_user(
|
||||
current_request: Request, verify_admin: bool
|
||||
) -> Tuple[Optional[HttpResponse], Optional[int]]:
|
||||
"""
|
||||
Return user id, if the provided token is valid and if user has admin
|
||||
rights if 'verify_admin' is True
|
||||
"""
|
||||
default_message = 'Provide a valid auth token.'
|
||||
auth_header = current_request.headers.get('Authorization')
|
||||
if not auth_header:
|
||||
@@ -98,9 +130,11 @@ def verify_user(current_request, verify_admin):
|
||||
return None, resp
|
||||
|
||||
|
||||
def authenticate(f):
|
||||
def authenticate(f: Callable) -> Callable:
|
||||
@wraps(f)
|
||||
def decorated_function(*args, **kwargs):
|
||||
def decorated_function(
|
||||
*args: Any, **kwargs: Any
|
||||
) -> Union[Callable, HttpResponse]:
|
||||
verify_admin = False
|
||||
response_object, resp = verify_user(request, verify_admin)
|
||||
if response_object:
|
||||
@@ -110,9 +144,11 @@ def authenticate(f):
|
||||
return decorated_function
|
||||
|
||||
|
||||
def authenticate_as_admin(f):
|
||||
def authenticate_as_admin(f: Callable) -> Callable:
|
||||
@wraps(f)
|
||||
def decorated_function(*args, **kwargs):
|
||||
def decorated_function(
|
||||
*args: Any, **kwargs: Any
|
||||
) -> Union[Callable, HttpResponse]:
|
||||
verify_admin = True
|
||||
response_object, resp = verify_user(request, verify_admin)
|
||||
if response_object:
|
||||
@@ -122,25 +158,36 @@ def authenticate_as_admin(f):
|
||||
return decorated_function
|
||||
|
||||
|
||||
def can_view_activity(auth_user_id, activity_user_id):
|
||||
def can_view_activity(
|
||||
auth_user_id: int, activity_user_id: int
|
||||
) -> Optional[HttpResponse]:
|
||||
"""
|
||||
Return error response if user has no right to view activity
|
||||
"""
|
||||
if auth_user_id != activity_user_id:
|
||||
return ForbiddenErrorResponse()
|
||||
return None
|
||||
|
||||
|
||||
def display_readable_file_size(size_in_bytes):
|
||||
def display_readable_file_size(size_in_bytes: Union[float, int]) -> str:
|
||||
"""
|
||||
Return readable file size from size in bytes
|
||||
"""
|
||||
if size_in_bytes == 0:
|
||||
return '0 bytes'
|
||||
if size_in_bytes == 1:
|
||||
return '1 byte'
|
||||
for unit in [' bytes', 'KB', 'MB', 'GB', 'TB']:
|
||||
if abs(size_in_bytes) < 1024.0:
|
||||
return f"{size_in_bytes:3.1f}{unit}"
|
||||
return f'{size_in_bytes:3.1f}{unit}'
|
||||
size_in_bytes /= 1024.0
|
||||
return f"{size_in_bytes} bytes"
|
||||
return f'{size_in_bytes} bytes'
|
||||
|
||||
|
||||
def get_readable_duration(duration, locale='en'):
|
||||
def get_readable_duration(duration: int, locale: Optional[str] = 'en') -> str:
|
||||
"""
|
||||
Return readable and localized duration from duration in seconds
|
||||
"""
|
||||
if locale is not None and locale != 'en':
|
||||
_t = humanize.i18n.activate(locale) # noqa
|
||||
readable_duration = humanize.naturaldelta(timedelta(seconds=duration))
|
||||
|
||||
Reference in New Issue
Block a user