first try at api using flask-restx & marshmallow

main
Alexander Graf 4 years ago
parent c30944404d
commit 866ad89dfc
No known key found for this signature in database
GPG Key ID: B8A9DC143E075629

@ -70,10 +70,12 @@ def create_app_from_config(config):
return utils.flask_babel.format_datetime(value) if value else '' return utils.flask_babel.format_datetime(value) if value else ''
# Import views # Import views
from mailu import ui, internal, sso from mailu import ui, internal, sso, api
app.register_blueprint(ui.ui, url_prefix=app.config['WEB_ADMIN']) app.register_blueprint(ui.ui, url_prefix=app.config['WEB_ADMIN'])
app.register_blueprint(internal.internal, url_prefix='/internal') app.register_blueprint(internal.internal, url_prefix='/internal')
app.register_blueprint(sso.sso, url_prefix='/sso') app.register_blueprint(sso.sso, url_prefix='/sso')
if app.config.get('API'):
api.register(app)
return app return app

@ -0,0 +1,38 @@
from flask import redirect, url_for
# import api version(s)
from . import v1
# api
ROOT='/api'
ACTIVE=v1
# patch url for swaggerui static assets
from flask_restx.apidoc import apidoc
apidoc.static_url_path = f'{ROOT}/swaggerui'
def register(app):
# register api bluprint(s)
app.register_blueprint(v1.blueprint, url_prefix=f'{ROOT}/v{int(v1.VERSION)}')
# add redirect to current api version
@app.route(f'{ROOT}/')
def redir():
return redirect(url_for(f'{ACTIVE.blueprint.name}.root'))
# swagger ui config
app.config.SWAGGER_UI_DOC_EXPANSION = 'list'
app.config.SWAGGER_UI_OPERATION_ID = True
app.config.SWAGGER_UI_REQUEST_DURATION = True
# TODO: remove patch of static assets for debugging
import os
if 'STATIC_ASSETS' in os.environ:
app.blueprints['ui'].static_folder = os.environ['STATIC_ASSETS']
# TODO: authentication via username + password
# TODO: authentication via api token
# TODO: api access for all users (via token)
# TODO: use permissions from "manager_of"
# TODO: switch to marshmallow, as parser is deprecated. use flask_accepts?

@ -0,0 +1,8 @@
from .. import models
def fqdn_in_use(*names):
for name in names:
for model in models.Domain, models.Alternative, models.Relay:
if model.query.get(name):
return model
return None

@ -0,0 +1,27 @@
from flask import Blueprint
from flask_restx import Api, fields
VERSION = 1.0
blueprint = Blueprint(f'api_v{int(VERSION)}', __name__)
api = Api(
blueprint, version=f'{VERSION:.1f}',
title='Mailu API', default_label='Mailu',
validate=True
)
response_fields = api.model('Response', {
'code': fields.Integer,
'message': fields.String,
})
error_fields = api.model('Error', {
'errors': fields.Nested(api.model('Error_Key', {
'key': fields.String,
'message':fields.String
})),
'message': fields.String,
})
from . import domains

@ -0,0 +1,183 @@
from flask_restx import Resource, fields, abort
from . import api, response_fields, error_fields
from .. import common
from ... import models
db = models.db
dom = api.namespace('domain', description='Domain operations')
alt = api.namespace('alternative', description='Alternative operations')
domain_fields = api.model('Domain', {
'name': fields.String(description='FQDN', example='example.com', required=True),
'comment': fields.String(description='a comment'),
'max_users': fields.Integer(description='maximum number of users', min=-1, default=-1),
'max_aliases': fields.Integer(description='maximum number of aliases', min=-1, default=-1),
'max_quota_bytes': fields.Integer(description='maximum quota for mailbox', min=0),
'signup_enabled': fields.Boolean(description='allow signup'),
# 'dkim_key': fields.String,
'alternatives': fields.List(fields.String(attribute='name', description='FQDN', example='example.com')),
})
# TODO - name ist required on creation but immutable on change
# TODO - name and alteranatives need to be checked to be a fqdn (regex)
domain_parser = api.parser()
domain_parser.add_argument('max_users', type=int, help='maximum number of users')
# TODO ... add more (or use marshmallow)
alternative_fields = api.model('Domain', {
'name': fields.String(description='alternative FQDN', example='example.com', required=True),
'domain': fields.String(description='domain FQDN', example='example.com', required=True),
'dkim_key': fields.String,
})
# TODO: domain and name are not always required and can't be changed
@dom.route('')
class Domains(Resource):
@dom.doc('list_domain')
@dom.marshal_with(domain_fields, as_list=True, skip_none=True, mask=['dkim_key'])
def get(self):
""" List domains """
return models.Domain.query.all()
@dom.doc('create_domain')
@dom.expect(domain_fields)
@dom.response(200, 'Success', response_fields)
@dom.response(400, 'Input validation exception', error_fields)
@dom.response(409, 'Duplicate domain name', error_fields)
def post(self):
""" Create a new domain """
data = api.payload
if common.fqdn_in_use(data['name']):
abort(409, f'Duplicate domain name {data["name"]!r}', errors={
'name': data['name'],
})
for item, created in models.Domain.from_dict(data):
if not created:
abort(409, f'Duplicate domain name {item.name!r}', errors={
'alternatives': item.name,
})
db.session.add(item)
db.session.commit()
@dom.route('/<name>')
class Domain(Resource):
@dom.doc('get_domain')
@dom.response(200, 'Success', domain_fields)
@dom.response(404, 'Domain not found')
@dom.marshal_with(domain_fields)
def get(self, name):
""" Find domain by name """
domain = models.Domain.query.get(name)
if not domain:
abort(404)
return domain
@dom.doc('update_domain')
@dom.expect(domain_fields)
@dom.response(200, 'Success', response_fields)
@dom.response(400, 'Input validation exception', error_fields)
@dom.response(404, 'Domain not found')
def put(self, name):
""" Update an existing domain """
domain = models.Domain.query.get(name)
if not domain:
abort(404)
data = api.payload
data['name'] = name
for item, created in models.Domain.from_dict(data):
if created is True:
db.session.add(item)
db.session.commit()
@dom.doc('modify_domain')
@dom.expect(domain_parser)
@dom.response(200, 'Success', response_fields)
@dom.response(400, 'Input validation exception', error_fields)
@dom.response(404, 'Domain not found')
def post(self, name=None):
""" Updates domain with form data """
domain = models.Domain.query.get(name)
if not domain:
abort(404)
data = dict(domain_parser.parse_args())
data['name'] = name
for item, created in models.Domain.from_dict(data):
if created is True:
db.session.add(item)
# TODO: flush?
db.session.commit()
@dom.doc('delete_domain')
@dom.response(200, 'Success', response_fields)
@dom.response(404, 'Domain not found')
def delete(self, name=None):
""" Delete domain """
domain = models.Domain.query.get(name)
if not domain:
abort(404)
db.session.delete(domain)
db.session.commit()
# @dom.route('/<name>/alternative')
# @alt.route('')
# class Alternatives(Resource):
# @alt.doc('alternatives_list')
# @alt.marshal_with(alternative_fields, as_list=True, skip_none=True, mask=['dkim_key'])
# def get(self, name=None):
# """ List alternatives (of domain) """
# if name is None:
# return models.Alternative.query.all()
# else:
# return models.Alternative.query.filter_by(domain_name = name).all()
# @alt.doc('alternative_create')
# @alt.expect(alternative_fields)
# @alt.response(200, 'Success', response_fields)
# @alt.response(400, 'Input validation exception', error_fields)
# @alt.response(404, 'Domain not found')
# @alt.response(409, 'Duplicate domain name', error_fields)
# def post(self, name=None):
# """ Create new alternative (for domain) """
# # abort(501)
# data = api.payload
# if name is not None:
# data['name'] = name
# domain = models.Domain.query.get(name)
# if not domain:
# abort(404)
# if common.fqdn_in_use(data['name']):
# abort(409, f'Duplicate domain name {data["name"]!r}', errors={
# 'name': data['name'],
# })
# for item, created in models.Alternative.from_dict(data):
# # TODO: handle creation of domain
# if not created:
# abort(409, f'Duplicate domain name {item.name!r}', errors={
# 'alternatives': item.name,
# })
# # db.session.add(item)
# # db.session.commit()
# @dom.route('/<name>/alternative/<alt>')
# @alt.route('/<name>')
# class Alternative(Resource):
# def get(self, name, alt=None):
# """ Find alternative (of domain) """
# abort(501)
# def put(self, name, alt=None):
# """ Update alternative (of domain) """
# abort(501)
# def post(self, name, alt=None):
# """ Update alternative (of domain) with form data """
# abort(501)
# def delete(self, name, alt=None):
# """ Delete alternative (for domain) """
# abort(501)
Loading…
Cancel
Save