221 lines
6.2 KiB
Python
221 lines
6.2 KiB
Python
"""Module containing operation related views."""
|
|
|
|
# vim: set tw=80 ts=4 sw=4 sts=4:
|
|
|
|
import dateutil.parser
|
|
|
|
from flask_jwt_extended import jwt_required
|
|
from flask_restplus import Namespace, Resource, fields
|
|
|
|
from ..models import db, result_as_dicts
|
|
from ..models.accounts import Account
|
|
from ..models.operations import Operation
|
|
|
|
|
|
# pylint: disable=invalid-name
|
|
ns = Namespace('operation', description='Operation management')
|
|
|
|
# Operation with sold model.
|
|
operation_model = ns.model('Operation', {
|
|
'id': fields.Integer(
|
|
default=None,
|
|
readonly=True,
|
|
description='Id of the operation'),
|
|
'operation_date': fields.Date(
|
|
dt_format='iso8601',
|
|
required=True,
|
|
description='Date of the operation'),
|
|
'label': fields.String(
|
|
required=True,
|
|
description='Label of the operation'),
|
|
'value': fields.Float(
|
|
required=True,
|
|
description='Value of the operation'),
|
|
'pointed': fields.Boolean(
|
|
required=True,
|
|
description='Pointed status of the operation'),
|
|
'category': fields.String(
|
|
required=False,
|
|
default=None,
|
|
description='Category of the operation'),
|
|
'account_id': fields.Integer(
|
|
required=True,
|
|
readonly=True,
|
|
description='Account id of the operation'),
|
|
'scheduled_operation_id': fields.Integer(
|
|
default=None,
|
|
readonly=True,
|
|
description='Scheduled operation ID of the operation'),
|
|
'confirmed': fields.Boolean(
|
|
description='Confirmed status of the operation'),
|
|
'canceled': fields.Boolean(
|
|
description='Canceled status of the operation (for a scheduled one)')
|
|
})
|
|
|
|
operation_with_balance_model = ns.clone(
|
|
'OperationWithBalance', operation_model, {
|
|
'balance': fields.Float(
|
|
readonly=True,
|
|
description='Cumulated balance'
|
|
),
|
|
}
|
|
)
|
|
|
|
# Parser for a date range and an account id.
|
|
account_range_parser = ns.parser()
|
|
account_range_parser.add_argument(
|
|
'begin',
|
|
type=lambda a: dateutil.parser.parse(a) if a else None,
|
|
required=False,
|
|
default=None,
|
|
location='args',
|
|
help='Begin date of the time period'
|
|
)
|
|
account_range_parser.add_argument(
|
|
'end',
|
|
type=lambda a: dateutil.parser.parse(a) if a else None,
|
|
required=False,
|
|
default=None,
|
|
location='args',
|
|
help='End date of the time period'
|
|
)
|
|
account_range_parser.add_argument(
|
|
'account_id',
|
|
type=int,
|
|
required=True,
|
|
location='args',
|
|
help='Id of the account'
|
|
)
|
|
# pylint: enable=invalid-name
|
|
|
|
|
|
# pylint: disable=no-self-use
|
|
@ns.route('/')
|
|
@ns.doc(
|
|
security='apikey',
|
|
responses={
|
|
401: 'Unauthorized'
|
|
})
|
|
class OperationListResource(Resource):
|
|
"""Resource to handle operation lists."""
|
|
|
|
@ns.response(200, 'OK', [operation_with_balance_model])
|
|
@ns.expect(account_range_parser)
|
|
@ns.marshal_list_with(operation_with_balance_model)
|
|
#@jwt_required
|
|
def get(self):
|
|
"""Get operations with cumulated balance for a specific account."""
|
|
|
|
data = account_range_parser.parse_args()
|
|
|
|
account_id = data['account_id']
|
|
begin = data['begin']
|
|
end = data['end']
|
|
|
|
return list(result_as_dicts(
|
|
Operation.get_with_balance(account_id, begin=begin, end=end)
|
|
)), 200
|
|
|
|
@ns.response(201, 'Operation created', operation_model)
|
|
@ns.response(404, 'Account not found')
|
|
@ns.response(406, 'Invalid operation data')
|
|
@ns.marshal_with(operation_model)
|
|
@jwt_required
|
|
def post(self):
|
|
"""Create a new operation."""
|
|
|
|
data = self.api.payload
|
|
|
|
account_id = data['account_id']
|
|
account = Account.query().get(account_id)
|
|
|
|
if not account:
|
|
ns.abort(404, 'Account with id %d not found.' % account_id)
|
|
|
|
# A new operation MUST NOT have an id;
|
|
if data.get('id') is not None:
|
|
ns.abort(406, 'Id must not be provided on creation.')
|
|
|
|
operation = Operation(**data)
|
|
|
|
db.session.add(operation) # pylint: disable=no-member
|
|
|
|
return operation, 201
|
|
|
|
|
|
@ns.route('/<int:operation_id>')
|
|
@ns.doc(
|
|
security='apikey',
|
|
params={
|
|
'operation_id': 'Id of the operation to manage'
|
|
},
|
|
responses={
|
|
401: 'Unauthorized',
|
|
404: 'Operation not found'
|
|
})
|
|
class OperationResource(Resource):
|
|
"""Resource to handle operations."""
|
|
|
|
@ns.response(200, 'OK', operation_model)
|
|
@ns.marshal_with(operation_model)
|
|
@jwt_required
|
|
def get(self, operation_id):
|
|
"""Get operation."""
|
|
|
|
# pylint: disable=no-member
|
|
operation = db.session.query(Operation).get(operation_id)
|
|
# pylint: enable=no-member
|
|
|
|
if not operation:
|
|
ns.abort(404, 'Operation with id %d not found.' % operation_id)
|
|
|
|
return operation, 200
|
|
|
|
@ns.expect(operation_model)
|
|
@ns.response(200, 'OK', operation_model)
|
|
@ns.response(406, 'Invalid operation data')
|
|
@ns.marshal_with(operation_model)
|
|
@jwt_required
|
|
def post(self, operation_id):
|
|
"""Update an operation."""
|
|
|
|
data = self.api.payload
|
|
|
|
# Check ID consistency.
|
|
if data.get('id', default=operation_id) != operation_id:
|
|
ns.abort(406, 'Id must not be provided or changed on update.')
|
|
|
|
# pylint: disable=no-member
|
|
operation = db.session.query(Operation).get(operation_id)
|
|
# pylint: enable=no-member
|
|
|
|
if not operation:
|
|
ns.abort(404, 'Operation with id %d not found.' % operation_id)
|
|
|
|
# FIXME check account_id consistency.
|
|
|
|
# SQLAlchemy objects ignore __dict__.update() with merge.
|
|
for key, value in data.items():
|
|
setattr(operation, key, value)
|
|
|
|
db.session.merge(operation) # pylint: disable=no-member
|
|
|
|
return operation, 200
|
|
|
|
@ns.response(204, 'Operation deleted', operation_model)
|
|
@ns.marshal_with(operation_model)
|
|
@jwt_required
|
|
def delete(self, operation_id):
|
|
"""Delete an operation."""
|
|
|
|
# pylint: disable=no-member
|
|
operation = db.session.query(Operation).get(operation_id)
|
|
# pylint: enable=no-member
|
|
|
|
if not operation:
|
|
ns.abort(404, 'Operation with id %d not found.' % operation_id)
|
|
|
|
db.session.delete(operation) # pylint: disable=no-member
|
|
|
|
return None, 204
|