Browse Source

Fix pydocstyle complaints

merge-requests/1/head
Drew Short 7 years ago
parent
commit
d2ddef20c8
  1. 69
      server/atheneum/__init__.py
  2. 4
      server/atheneum/api/__init__.py
  3. 22
      server/atheneum/api/authentication_api.py
  4. 12
      server/atheneum/api/decorators.py
  5. 10
      server/atheneum/api/model.py
  6. 10
      server/atheneum/db.py
  7. 4
      server/atheneum/default_settings.py
  8. 1
      server/atheneum/middleware/__init__.py
  9. 22
      server/atheneum/middleware/authentication_middleware.py
  10. 4
      server/atheneum/model/__init__.py
  11. 16
      server/atheneum/model/user_model.py
  12. 1
      server/atheneum/service/__init__.py
  13. 48
      server/atheneum/service/authentication_service.py
  14. 28
      server/atheneum/service/user_service.py
  15. 25
      server/atheneum/service/user_token_service.py
  16. 1
      server/atheneum/utility/__init__.py
  17. 14
      server/atheneum/utility/authentication_utility.py
  18. 12
      server/atheneum/utility/json_utility.py
  19. 6
      server/manage.py
  20. 19
      server/tests/conftest.py

69
server/atheneum/__init__.py

@ -1,16 +1,12 @@
"""
Atheneum Flask Application
"""
"""Atheneum Flask Application."""
import os
from logging.config import dictConfig
from flask import Flask
from flask_migrate import Migrate, upgrade
from flask_sqlalchemy import SQLAlchemy
from atheneum import utility
from flask_migrate import Migrate
db: SQLAlchemy = SQLAlchemy()
from atheneum.db import db
from atheneum.utility import json_utility
dictConfig({
'version': 1,
@ -31,69 +27,64 @@ dictConfig({
def create_app(test_config: dict = None) -> Flask:
"""
Create an instance of Atheneum
Create an instance of Atheneum.
:param test_config:
:return:
"""
atheneum_app = Flask(__name__, instance_relative_config=True)
logger = atheneum_app.logger()
logger.debug('Creating Atheneum Server')
app = Flask(__name__, instance_relative_config=True)
app.logger.debug('Creating Atheneum Server')
data_directory = os.getenv('ATHENEUM_DATA_DIRECTORY', '/tmp')
logger.debug('Atheneum Data Directory: %s', data_directory)
app.logger.debug('Atheneum Data Directory: %s', data_directory)
default_database_uri = 'sqlite:///{}/atheneum.db'.format(data_directory)
atheneum_app.config.from_mapping(
app.config.from_mapping(
SECRET_KEY='dev',
SQLALCHEMY_DATABASE_URI=default_database_uri,
SQLALCHEMY_TRACK_MODIFICATIONS=False
)
if test_config is None:
logger.debug('Loading configurations')
atheneum_app.config.from_object('atheneum.default_settings')
atheneum_app.config.from_pyfile('config.py', silent=True)
app.logger.debug('Loading configurations')
app.config.from_object('atheneum.default_settings')
app.config.from_pyfile('config.py', silent=True)
if os.getenv('ATHENEUM_SETTINGS', None):
atheneum_app.config.from_envvar('ATHENEUM_SETTINGS')
app.config.from_envvar('ATHENEUM_SETTINGS')
else:
logger.debug('Loading test configuration')
atheneum_app.config.from_object(test_config)
app.logger.debug('Loading test configuration')
app.config.from_object(test_config)
try:
os.makedirs(atheneum_app.instance_path)
os.makedirs(app.instance_path)
except OSError:
pass
atheneum_app.json_encoder = utility.CustomJSONEncoder
app.json_encoder = json_utility.CustomJSONEncoder
logger.debug('Initializing Application')
db.init_app(atheneum_app)
logger.debug('Registering Database Models')
Migrate(atheneum_app, db)
app.logger.debug('Initializing Application')
db.init_app(app)
app.logger.debug('Registering Database Models')
Migrate(app, db)
return atheneum_app
return app
def register_blueprints(atheneum_app: Flask) -> None:
def register_blueprints(app: Flask) -> None:
"""
Register blueprints for the application
Register blueprints for the application.
:param atheneum_app:
:param app:
:return:
"""
from atheneum.api import AUTH_BLUEPRINT
atheneum_app.register_blueprint(AUTH_BLUEPRINT)
APP = create_app()
register_blueprints(APP)
app.register_blueprint(AUTH_BLUEPRINT)
def init_db() -> None:
"""Clear existing data and create new tables."""
upgrade('migrations')
atheneum = create_app() # pylint: disable=C0103
register_blueprints(atheneum)
logger = atheneum.logger # pylint: disable=C0103
if __name__ == "__main__":
APP.run()
atheneum.run()

4
server/atheneum/api/__init__.py

@ -1,4 +1,2 @@
"""
API blueprint exports
"""
"""API blueprint exports."""
from atheneum.api.authentication_api import AUTH_BLUEPRINT

22
server/atheneum/api/authentication_api.py

@ -1,13 +1,14 @@
"""
Authentication API blueprint and endpoint definitions
"""
"""Authentication API blueprint and endpoint definitions."""
from flask import Blueprint, g
from atheneum.api.decorators import return_json
from atheneum.api.model import APIResponse
from atheneum.middleware import authentication_middleware
from atheneum.service import user_token_service, authentication_service
from atheneum.service import (
user_token_service,
authentication_service,
user_service
)
AUTH_BLUEPRINT = Blueprint(
name='auth', import_name=__name__, url_prefix='/auth')
@ -18,7 +19,8 @@ AUTH_BLUEPRINT = Blueprint(
@authentication_middleware.require_basic_auth
def login() -> APIResponse:
"""
Get a token for continued authentication
Get a token for continued authentication.
:return: A login token for continued authentication
"""
user_token = user_token_service.create(g.user)
@ -30,10 +32,11 @@ def login() -> APIResponse:
@authentication_middleware.require_token_auth
def login_bump() -> APIResponse:
"""
Update the user last seen timestamp
Update the user last seen timestamp.
:return: A time stamp for the bumped login
"""
authentication_service.bump_login(g.user)
user_service.update_last_login_time(g.user)
return APIResponse({'last_login_time': g.user.last_login_time}, 200)
@ -42,7 +45,8 @@ def login_bump() -> APIResponse:
@authentication_middleware.require_token_auth
def logout() -> APIResponse:
"""
logout and delete a token
Logout and delete a token.
:return:
"""
authentication_service.logout(g.user_token)

12
server/atheneum/api/decorators.py

@ -1,7 +1,4 @@
"""
Decorators to be used in the api module
"""
"""Decorators to be used in the api module."""
from functools import wraps
from typing import Callable, Any
@ -12,15 +9,16 @@ from atheneum.api.model import APIResponse
def return_json(func: Callable) -> Callable:
"""
If an Response object is not returned, jsonify the result and return it
If an Response object is not returned, jsonify the result and return it.
:param func:
:return:
"""
@wraps(func)
def decorate(*args: list, **kwargs: dict) -> Any:
"""
Make sure that the return of the function is jsonified into a Response
Make sure that the return of the function is jsonified into a Response.
:param args:
:param kwargs:
:return:

10
server/atheneum/api/model.py

@ -1,13 +1,9 @@
"""
Model definitions for the api module
"""
"""Model definitions for the api module."""
from typing import Any, NamedTuple
class APIResponse(NamedTuple): # pylint: disable=too-few-public-methods
"""
Custom class to wrap api responses
"""
"""Custom class to wrap api responses."""
payload: Any
status: int

10
server/atheneum/db.py

@ -0,0 +1,10 @@
"""Database configuration and methods."""
from flask_migrate import upgrade
from flask_sqlalchemy import SQLAlchemy
db: SQLAlchemy = SQLAlchemy()
def init_db() -> None:
"""Clear existing data and create new tables."""
upgrade('migrations')

4
server/atheneum/default_settings.py

@ -1,6 +1,4 @@
"""
Default settings for atheneum
"""
"""Default settings for atheneum."""
DEBUG = False
SECRET_KEY = b'\xb4\x89\x0f\x0f\xe5\x88\x97\xfe\x8d<\x0b@d\xe9\xa5\x87%' \

1
server/atheneum/middleware/__init__.py

@ -0,0 +1 @@
"""Middleware package."""

22
server/atheneum/middleware/authentication_middleware.py

@ -1,6 +1,4 @@
"""
Middleware to handle authentication
"""
"""Middleware to handle authentication."""
import base64
from functools import wraps
from typing import Optional, Callable, Any
@ -19,7 +17,7 @@ from atheneum.service import (
def authenticate_with_password(name: str, password: str) -> bool:
"""
Authenticate a username and a password
Authenticate a username and a password.
:param name:
:param password:
@ -35,7 +33,7 @@ def authenticate_with_password(name: str, password: str) -> bool:
def authenticate_with_token(name: str, token: str) -> bool:
"""
Authenticate a username and a token
Authenticate a username and a token.
:param name:
:param token:
@ -54,7 +52,7 @@ def authenticate_with_token(name: str, token: str) -> bool:
def authentication_failed(auth_type: str) -> Response:
"""
Return a correct response for failed authentication
Return a correct response for failed authentication.
:param auth_type:
:return:
@ -69,7 +67,7 @@ def authentication_failed(auth_type: str) -> Response:
def parse_token_header(
header_value: str) -> Optional[Authorization]:
"""
Parse the Authorization: Token header for the username and token
Parse the Authorization: Token header for the username and token.
:param header_value:
:return:
@ -94,7 +92,7 @@ def parse_token_header(
def require_basic_auth(func: Callable) -> Callable:
"""
Decorator to require inline basic auth
Decorate require inline basic auth.
:param func:
:return:
@ -102,8 +100,7 @@ def require_basic_auth(func: Callable) -> Callable:
@wraps(func)
def decorate(*args: list, **kwargs: dict) -> Any:
"""
authenticate with password from basic auth before calling wrapped
function
Authenticate with a password.
:param args:
:param kwargs:
@ -119,7 +116,7 @@ def require_basic_auth(func: Callable) -> Callable:
def require_token_auth(func: Callable) -> Callable:
"""
Decorator to require inline token auth
Decorate require inline token auth.
:param func:
:return:
@ -127,8 +124,7 @@ def require_token_auth(func: Callable) -> Callable:
@wraps(func)
def decorate(*args: list, **kwargs: dict) -> Any:
"""
Authenticate with a token from token auth before calling wrapped
function
Authenticate with a token.
:param args:
:param kwargs:

4
server/atheneum/model/__init__.py

@ -1,4 +1,2 @@
"""
Expose models to be used in Atheneum
"""
"""Expose models to be used in Atheneum."""
from atheneum.model.user_model import User, UserToken

16
server/atheneum/model/user_model.py

@ -1,13 +1,10 @@
"""
User related SQLALchemy models
"""
from atheneum import db
"""User related SQLALchemy models."""
from atheneum.db import db
class User(db.Model): # pylint: disable=too-few-public-methods
"""
Represents a user in the system
"""
"""Represents a user in the system."""
__tablename__ = 'user'
ROLE_USER = 'USER'
@ -29,9 +26,8 @@ class User(db.Model): # pylint: disable=too-few-public-methods
class UserToken(db.Model): # pylint: disable=too-few-public-methods
"""
Represents a token used alongside a user to authenticate operations
"""
"""Represents a token used alongside a user to authenticate operations."""
__tablename__ = 'user_token'
user_token_id = db.Column('id', db.Integer, primary_key=True)

1
server/atheneum/service/__init__.py

@ -0,0 +1 @@
"""Service package."""

48
server/atheneum/service/authentication_service.py

@ -1,38 +1,17 @@
"""
Service to handle authentication
"""
import uuid
"""Service to handle authentication."""
from datetime import datetime
from typing import Optional, Tuple
from typing import Optional
from nacl import pwhash
from nacl.exceptions import InvalidkeyError
from atheneum.model import User, UserToken
from atheneum.service import user_service, user_token_service
def generate_token() -> uuid.UUID:
"""
Generate a unique token
:return:
"""
return uuid.uuid4()
def get_password_hash(password: str) -> Tuple[str, int]:
"""
Retrieve argon2id password hash.
:param password: plaintext password to convert
:return: Tuple[password_hash, password_revision]
"""
return pwhash.argon2id.str(password.encode('utf8')).decode('utf8'), 1
from atheneum.service import user_token_service
def is_valid_password(user: User, password: str) -> bool:
"""
User password must pass pwhash verify
User password must pass pwhash verify.
:param user:
:param password:
@ -50,8 +29,10 @@ def is_valid_password(user: User, password: str) -> bool:
def is_valid_token(user_token: Optional[UserToken]) -> bool:
"""
Token must be enabled and if it has an expiration, it must be
greater than now.
Validate a token.
Token must be enabled and if it has an expiration, it must be greater
than now.
:param user_token:
:return:
@ -66,20 +47,9 @@ def is_valid_token(user_token: Optional[UserToken]) -> bool:
return True
def bump_login(user: Optional[User]) -> None:
"""
Update the last login time for the user
:param user:
:return:
"""
if user is not None:
user_service.update_last_login_time(user)
def logout(user_token: Optional[UserToken] = None) -> None:
"""
Remove a user_token associated with a client session
Remove a user_token associated with a client session.
:param user_token:
:return:

28
server/atheneum/service/user_service.py

@ -1,24 +1,25 @@
"""
Service to handle user operations
"""
"""Service to handle user operations."""
import logging
from datetime import datetime
from typing import Optional
from atheneum import APP, db
from atheneum.db import db
from atheneum.model import User
from atheneum.service import authentication_service
from atheneum.utility import authentication_utility
LOGGER = logging.getLogger(__name__)
def register(name: str, password: str, role: str) -> User:
"""
Register a new user
Register a new user.
:param name: Desired user name. Must be unique and not already registered
:param password: Password to be hashed and stored for the user
:param role: Role to assign the user [ROLE_USER, ROLE_ADMIN]
:return:
"""
pw_hash, pw_revision = authentication_service.get_password_hash(password)
pw_hash, pw_revision = authentication_utility.get_password_hash(password)
new_user = User(
name=name,
@ -30,13 +31,13 @@ def register(name: str, password: str, role: str) -> User:
db.session.add(new_user)
db.session.commit()
APP.logger.info('Registered new user: %s with role: %s', name, role)
LOGGER.info('Registered new user: %s with role: %s', name, role)
return new_user
def delete(user: User) -> bool:
"""
Delete a user
Delete a user.
:param user:
:return:
@ -50,24 +51,25 @@ def delete(user: User) -> bool:
def update_last_login_time(user: User) -> None:
"""
Bump the last login time for the user
Bump the last login time for the user.
:param user:
:return:
"""
if user is not None:
user.last_login_time = datetime.now()
db.session.commit()
def update_password(user: User, password: str) -> None:
"""
Change the user password
Change the user password.
:param user:
:param password:
:return:
"""
pw_hash, pw_revision = authentication_service.get_password_hash(
pw_hash, pw_revision = authentication_utility.get_password_hash(
password)
user.password_hash = pw_hash
user.password_revision = pw_revision
@ -76,7 +78,7 @@ def update_password(user: User, password: str) -> None:
def find_by_name(name: str) -> Optional[User]:
"""
Find a user by name
Find a user by name.
:param name:
:return:

25
server/atheneum/service/user_token_service.py

@ -1,12 +1,19 @@
"""
Service to handle user_token operations
"""
"""Service to handle user_token operations."""
import uuid
from datetime import datetime
from typing import Optional
from atheneum import db
from atheneum.db import db
from atheneum.model import User, UserToken
from atheneum.service import authentication_service
def generate_token() -> uuid.UUID:
"""
Generate a unique token.
:return:
"""
return uuid.uuid4()
def create(
@ -15,7 +22,7 @@ def create(
enabled: bool = True,
expiration_time: Optional[datetime] = None) -> UserToken:
"""
Create and save a UserToken
Create and save a UserToken.
:param user: The User object to bind the token to
:param note: An optional field to store additional information about a
@ -27,7 +34,7 @@ def create(
no expiration
:return:
"""
token = authentication_service.generate_token()
token = generate_token()
user_token = UserToken(
user_id=user.id,
token=token.__str__(),
@ -45,7 +52,7 @@ def create(
def delete(user_token: UserToken) -> bool:
"""
Delete a user_token
Delete a user_token.
:param user_token:
:return:
@ -59,7 +66,7 @@ def delete(user_token: UserToken) -> bool:
def find_by_user_and_token(user: User, token: str) -> Optional[UserToken]:
"""
Lookup a user_token by user and token string
Lookup a user_token by user and token string.
:param user:
:param token:

1
server/atheneum/utility/__init__.py

@ -0,0 +1 @@
"""Utilities for Atheneum."""

14
server/atheneum/utility/authentication_utility.py

@ -0,0 +1,14 @@
"""Authentication specific utilities."""
from typing import Tuple
from nacl import pwhash
def get_password_hash(password: str) -> Tuple[str, int]:
"""
Retrieve argon2id password hash.
:param password: plaintext password to convert
:return: Tuple[password_hash, password_revision]
"""
return pwhash.argon2id.str(password.encode('utf8')).decode('utf8'), 1

12
server/atheneum/utility.py → server/atheneum/utility/json_utility.py

@ -1,8 +1,4 @@
"""
Utility classes for Atheneum
"""
"""JSON specific utilities."""
from datetime import date
from typing import Any
@ -11,10 +7,10 @@ from flask.json import JSONEncoder
class CustomJSONEncoder(JSONEncoder):
"""
Ensure that datetime values are serialized correctly
"""
"""Ensure that datetime values are serialized correctly."""
def default(self, o: Any) -> Any: # pylint: disable=E0202
"""Handle encoding date and datetime objects according to rfc3339."""
try:
if isinstance(o, date):
return rfc3339.format(o)

6
server/manage.py

@ -7,7 +7,7 @@ from os import path
import click
from click import Context
from atheneum import APP
from atheneum import atheneum
from atheneum.model import User
from atheneum.service import user_service
@ -119,6 +119,6 @@ user_command_group.add_command(reset_user_password)
user_command_group.add_command(list_users)
if __name__ == '__main__':
logging.debug('Managing: %s', APP.name)
with APP.app_context():
logging.debug('Managing: %s', atheneum.name)
with atheneum.app_context():
main()

19
server/tests/conftest.py

@ -10,7 +10,8 @@ from flask import Flask
from flask.testing import FlaskClient, FlaskCliRunner
from werkzeug.test import Client
from atheneum import create_app, init_db, register_blueprints
from atheneum import create_app, register_blueprints
from atheneum.db import init_db
from atheneum.model import User
from atheneum.service import user_service
@ -26,25 +27,25 @@ def add_test_user() -> Tuple[str, str]:
@pytest.fixture
def app() -> Flask:
"""Create and configure a new app instance for each test."""
"""Create and configure a new atheneum_app instance for each test."""
# create a temporary file to isolate the database for each test
db_fd, db_path = tempfile.mkstemp()
# create the app with common test config
app = create_app({
# create the atheneum_app with common test config
atheneum_app = create_app({
'TESTING': True,
'DATABASE': db_path,
})
register_blueprints(app)
register_blueprints(atheneum_app)
# create the database and load test data
with app.app_context():
with atheneum_app.app_context():
init_db()
test_username, test_password = add_test_user()
app.config['test_username'] = test_username
app.config['test_password'] = test_password
atheneum_app.config['test_username'] = test_username
atheneum_app.config['test_password'] = test_password
# get_db().executescript(_data_sql)
yield app
yield atheneum_app
# close and remove the temporary database
os.close(db_fd)

Loading…
Cancel
Save