mirror of
https://github.com/dancojocaru2000/foxbank.git
synced 2025-06-19 11:02:28 +03:00
Compare commits
No commits in common. "5e19ab457ba87e567e0787d6841458968461a4e3" and "e369315034dd70677dc9d2e9ef2decdad830e25e" have entirely different histories.
5e19ab457b
...
e369315034
26 changed files with 200 additions and 1277 deletions
|
@ -10,15 +10,5 @@
|
||||||
"path": "."
|
"path": "."
|
||||||
}
|
}
|
||||||
],
|
],
|
||||||
"settings": {
|
"settings": {}
|
||||||
"sqltools.connections": [
|
|
||||||
{
|
|
||||||
"previewLimit": 50,
|
|
||||||
"driver": "SQLite",
|
|
||||||
"database": "${workspaceFolder:server}/data/db.sqlite",
|
|
||||||
"name": "Server DB"
|
|
||||||
}
|
|
||||||
],
|
|
||||||
"sqltools.useNodeRuntime": true
|
|
||||||
}
|
|
||||||
}
|
}
|
6
server/.vscode/launch.json
vendored
6
server/.vscode/launch.json
vendored
|
@ -11,15 +11,13 @@
|
||||||
"module": "flask",
|
"module": "flask",
|
||||||
"env": {
|
"env": {
|
||||||
"FLASK_APP": "server.py",
|
"FLASK_APP": "server.py",
|
||||||
"FLASK_ENV": "development",
|
"FLASK_ENV": "development"
|
||||||
"FLASK_RUN_PORT": "5001"
|
|
||||||
},
|
},
|
||||||
"args": [
|
"args": [
|
||||||
"run",
|
"run",
|
||||||
"--no-debugger"
|
"--no-debugger"
|
||||||
],
|
],
|
||||||
"jinja": true,
|
"jinja": true
|
||||||
"justMyCode": false
|
|
||||||
}
|
}
|
||||||
]
|
]
|
||||||
}
|
}
|
|
@ -8,7 +8,6 @@ flask = "*"
|
||||||
gunicorn = "*"
|
gunicorn = "*"
|
||||||
pyotp = "*"
|
pyotp = "*"
|
||||||
flask-cors = "*"
|
flask-cors = "*"
|
||||||
flask-smorest = "*"
|
|
||||||
|
|
||||||
[dev-packages]
|
[dev-packages]
|
||||||
|
|
||||||
|
|
45
server/Pipfile.lock
generated
45
server/Pipfile.lock
generated
|
@ -1,7 +1,7 @@
|
||||||
{
|
{
|
||||||
"_meta": {
|
"_meta": {
|
||||||
"hash": {
|
"hash": {
|
||||||
"sha256": "b70c68cd833afb9cc5b924eed2688784766705c1d4e008302bcc93d05f6bdbd4"
|
"sha256": "2ba252d63658abd009170d14705593521c57c99f82b643fcf232eeb51be35d10"
|
||||||
},
|
},
|
||||||
"pipfile-spec": 6,
|
"pipfile-spec": 6,
|
||||||
"requires": {
|
"requires": {
|
||||||
|
@ -16,17 +16,6 @@
|
||||||
]
|
]
|
||||||
},
|
},
|
||||||
"default": {
|
"default": {
|
||||||
"apispec": {
|
|
||||||
"extras": [
|
|
||||||
"marshmallow"
|
|
||||||
],
|
|
||||||
"hashes": [
|
|
||||||
"sha256:5bc5404b19259aeeb307ce9956e2c1a97722c6a130ef414671dfc21acd622afc",
|
|
||||||
"sha256:d167890e37f14f3f26b588ff2598af35faa5c27612264ea1125509c8ff860834"
|
|
||||||
],
|
|
||||||
"markers": "python_version >= '3.6'",
|
|
||||||
"version": "==5.1.1"
|
|
||||||
},
|
|
||||||
"click": {
|
"click": {
|
||||||
"hashes": [
|
"hashes": [
|
||||||
"sha256:353f466495adaeb40b6b5f592f9f91cb22372351c84caeb068132442a4518ef3",
|
"sha256:353f466495adaeb40b6b5f592f9f91cb22372351c84caeb068132442a4518ef3",
|
||||||
|
@ -51,14 +40,6 @@
|
||||||
"index": "pypi",
|
"index": "pypi",
|
||||||
"version": "==3.0.10"
|
"version": "==3.0.10"
|
||||||
},
|
},
|
||||||
"flask-smorest": {
|
|
||||||
"hashes": [
|
|
||||||
"sha256:b08b20fb15e505f4f032a82dd0d5471e431d1b8da9ae16e4a0099bb70d753c47",
|
|
||||||
"sha256:d97e114b972a0afae6a6c7883069c753c425ae1d8bb4c548028536465b1d1e19"
|
|
||||||
],
|
|
||||||
"index": "pypi",
|
|
||||||
"version": "==0.35.0"
|
|
||||||
},
|
|
||||||
"gunicorn": {
|
"gunicorn": {
|
||||||
"hashes": [
|
"hashes": [
|
||||||
"sha256:9dcc4547dbb1cb284accfb15ab5667a0e5d1881cc443e0677b4882a4067a807e",
|
"sha256:9dcc4547dbb1cb284accfb15ab5667a0e5d1881cc443e0677b4882a4067a807e",
|
||||||
|
@ -158,14 +139,6 @@
|
||||||
"markers": "python_version >= '3.6'",
|
"markers": "python_version >= '3.6'",
|
||||||
"version": "==2.0.1"
|
"version": "==2.0.1"
|
||||||
},
|
},
|
||||||
"marshmallow": {
|
|
||||||
"hashes": [
|
|
||||||
"sha256:04438610bc6dadbdddb22a4a55bcc7f6f8099e69580b2e67f5a681933a1f4400",
|
|
||||||
"sha256:4c05c1684e0e97fe779c62b91878f173b937fe097b356cd82f793464f5bc6138"
|
|
||||||
],
|
|
||||||
"markers": "python_version >= '3.6'",
|
|
||||||
"version": "==3.14.1"
|
|
||||||
},
|
|
||||||
"pyotp": {
|
"pyotp": {
|
||||||
"hashes": [
|
"hashes": [
|
||||||
"sha256:9d144de0f8a601d6869abe1409f4a3f75f097c37b50a36a3bf165810a6e23f28",
|
"sha256:9d144de0f8a601d6869abe1409f4a3f75f097c37b50a36a3bf165810a6e23f28",
|
||||||
|
@ -176,11 +149,11 @@
|
||||||
},
|
},
|
||||||
"setuptools": {
|
"setuptools": {
|
||||||
"hashes": [
|
"hashes": [
|
||||||
"sha256:10d6eff7fc27ada30cc87e21abf324713b7169b97af1f81f8744d66260e91d10",
|
"sha256:6d10741ff20b89cd8c6a536ee9dc90d3002dec0226c78fb98605bfb9ef8a7adf",
|
||||||
"sha256:89e8cb2d5ade19e9885e56cd110f2f1e80697f7cffa048886c585fe559ebbe32"
|
"sha256:d144f85102f999444d06f9c0e8c737fd0194f10f2f7e5fdb77573f6e2fa4fad0"
|
||||||
],
|
],
|
||||||
"markers": "python_version >= '3.7'",
|
"markers": "python_version >= '3.6'",
|
||||||
"version": "==60.1.1"
|
"version": "==59.5.0"
|
||||||
},
|
},
|
||||||
"six": {
|
"six": {
|
||||||
"hashes": [
|
"hashes": [
|
||||||
|
@ -190,14 +163,6 @@
|
||||||
"markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'",
|
"markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3'",
|
||||||
"version": "==1.16.0"
|
"version": "==1.16.0"
|
||||||
},
|
},
|
||||||
"webargs": {
|
|
||||||
"hashes": [
|
|
||||||
"sha256:bb3530b0d37cdc5a5e29d30034dde4351811b9bc345eef21eb070a3ea7562093",
|
|
||||||
"sha256:bcce022250ee97cfbb0ad07b02388ac90a226ef4b479ec84317152345a565614"
|
|
||||||
],
|
|
||||||
"markers": "python_version >= '3.6'",
|
|
||||||
"version": "==8.0.1"
|
|
||||||
},
|
|
||||||
"werkzeug": {
|
"werkzeug": {
|
||||||
"hashes": [
|
"hashes": [
|
||||||
"sha256:63d3dc1cf60e7b7e35e97fa9861f7397283b75d765afcaefd993d6046899de8f",
|
"sha256:63d3dc1cf60e7b7e35e97fa9861f7397283b75d765afcaefd993d6046899de8f",
|
||||||
|
|
|
@ -2,12 +2,10 @@ import sqlite3
|
||||||
|
|
||||||
from flask import current_app, g
|
from flask import current_app, g
|
||||||
|
|
||||||
import os
|
DB_FILE = './data/db.sqlite'
|
||||||
DB_FILE = os.environ.get('DB_FILE', './data/db.sqlite')
|
|
||||||
|
|
||||||
get_return = sqlite3.Connection
|
get_return = sqlite3.Connection
|
||||||
|
|
||||||
|
|
||||||
def get() -> get_return:
|
def get() -> get_return:
|
||||||
if 'db' not in g:
|
if 'db' not in g:
|
||||||
g.db = sqlite3.connect(
|
g.db = sqlite3.connect(
|
||||||
|
@ -18,14 +16,12 @@ def get() -> get_return:
|
||||||
|
|
||||||
return g.db
|
return g.db
|
||||||
|
|
||||||
|
|
||||||
def close(e=None):
|
def close(e=None):
|
||||||
db = g.pop('db', None)
|
db = g.pop('db', None)
|
||||||
|
|
||||||
if db:
|
if db:
|
||||||
db.close()
|
db.close()
|
||||||
|
|
||||||
|
|
||||||
def init():
|
def init():
|
||||||
db = get()
|
db = get()
|
||||||
|
|
||||||
|
@ -33,7 +29,6 @@ def init():
|
||||||
db.executescript(f.read().decode('utf8'))
|
db.executescript(f.read().decode('utf8'))
|
||||||
db.commit()
|
db.commit()
|
||||||
|
|
||||||
|
|
||||||
def init_app(app):
|
def init_app(app):
|
||||||
app.teardown_appcontext(close)
|
app.teardown_appcontext(close)
|
||||||
|
|
24
server/db_utils.py
Normal file
24
server/db_utils.py
Normal file
|
@ -0,0 +1,24 @@
|
||||||
|
from functools import wraps
|
||||||
|
|
||||||
|
import db
|
||||||
|
import models
|
||||||
|
|
||||||
|
def get_db(fn):
|
||||||
|
@wraps(fn)
|
||||||
|
def wrapper(*args, **kargs):
|
||||||
|
return fn(db.get(), *args, **kargs)
|
||||||
|
return wrapper
|
||||||
|
|
||||||
|
@get_db
|
||||||
|
def get_user(db: db.get_return, username: str|None = None, user_id: int|None = None) -> models.User | None:
|
||||||
|
cur = db.cursor()
|
||||||
|
if username is not None:
|
||||||
|
cur.execute('select * from users where username=?', (username,))
|
||||||
|
elif user_id is not None:
|
||||||
|
cur.execute('select * from users where id=?', (user_id,))
|
||||||
|
else:
|
||||||
|
raise Exception('Neither username or user_id passed')
|
||||||
|
result = cur.fetchone()
|
||||||
|
if result is None:
|
||||||
|
return None
|
||||||
|
return models.User.from_query(result)
|
12
server/decorators.py
Normal file
12
server/decorators.py
Normal file
|
@ -0,0 +1,12 @@
|
||||||
|
from http import HTTPStatus
|
||||||
|
from functools import wraps
|
||||||
|
|
||||||
|
def no_content(fn):
|
||||||
|
@wraps(fn)
|
||||||
|
def wrapper(*args, **kargs):
|
||||||
|
result = fn(*args, **kargs)
|
||||||
|
if result is None:
|
||||||
|
return None, HTTPStatus.NO_CONTENT
|
||||||
|
else:
|
||||||
|
return result
|
||||||
|
return wrapper
|
|
@ -1,33 +0,0 @@
|
||||||
from flask import Flask
|
|
||||||
from .apis import init_apis
|
|
||||||
|
|
||||||
class Config:
|
|
||||||
OPENAPI_VERSION = "3.0.2"
|
|
||||||
OPENAPI_JSON_PATH = "api-spec.json"
|
|
||||||
OPENAPI_URL_PREFIX = "/"
|
|
||||||
OPENAPI_REDOC_PATH = "/redoc"
|
|
||||||
OPENAPI_REDOC_URL = (
|
|
||||||
"https://cdn.jsdelivr.net/npm/redoc@next/bundles/redoc.standalone.js"
|
|
||||||
)
|
|
||||||
OPENAPI_SWAGGER_UI_PATH = "/swagger-ui"
|
|
||||||
OPENAPI_SWAGGER_UI_URL = "https://cdn.jsdelivr.net/npm/swagger-ui-dist/"
|
|
||||||
OPENAPI_RAPIDOC_PATH = "/rapidoc"
|
|
||||||
OPENAPI_RAPIDOC_URL = "https://unpkg.com/rapidoc/dist/rapidoc-min.js"
|
|
||||||
|
|
||||||
def create_app():
|
|
||||||
app = Flask(__name__)
|
|
||||||
app.config.from_object(Config())
|
|
||||||
|
|
||||||
init_db(app)
|
|
||||||
init_cors(app)
|
|
||||||
init_apis(app)
|
|
||||||
|
|
||||||
return app
|
|
||||||
|
|
||||||
def init_cors(app):
|
|
||||||
from flask_cors import CORS
|
|
||||||
cors = CORS(app)
|
|
||||||
|
|
||||||
def init_db(app):
|
|
||||||
from .db import init_app
|
|
||||||
init_app(app)
|
|
|
@ -1,33 +0,0 @@
|
||||||
from flask import Flask
|
|
||||||
from flask_smorest import Api
|
|
||||||
|
|
||||||
from .accounts import bp as acc_bp
|
|
||||||
from .login import bp as login_bp
|
|
||||||
from .transactions import bp as transactions_bp
|
|
||||||
from .notifications import bp as notifications_bp
|
|
||||||
|
|
||||||
class ApiWithErr(Api):
|
|
||||||
def handle_http_exception(self, error):
|
|
||||||
if error.data and error.data['response']:
|
|
||||||
return error.data['response']
|
|
||||||
return super().handle_http_exception(error)
|
|
||||||
|
|
||||||
def init_apis(app: Flask):
|
|
||||||
api = ApiWithErr(app, spec_kwargs={
|
|
||||||
'title': 'FoxBank',
|
|
||||||
'version': '1',
|
|
||||||
'openapi_version': '3.0.0',
|
|
||||||
'components': {
|
|
||||||
'securitySchemes': {
|
|
||||||
'Token': {
|
|
||||||
'type': 'http',
|
|
||||||
'scheme': 'bearer',
|
|
||||||
'bearerFormat': 'Token ',
|
|
||||||
}
|
|
||||||
}
|
|
||||||
},
|
|
||||||
})
|
|
||||||
api.register_blueprint(login_bp, url_prefix='/login')
|
|
||||||
api.register_blueprint(acc_bp, url_prefix='/accounts')
|
|
||||||
api.register_blueprint(transactions_bp, url_prefix='/transactions')
|
|
||||||
api.register_blueprint(notifications_bp, url_prefix='/notifications')
|
|
|
@ -1,141 +0,0 @@
|
||||||
from http import HTTPStatus
|
|
||||||
from flask.views import MethodView
|
|
||||||
from flask_smorest import Blueprint, abort
|
|
||||||
from marshmallow import Schema, fields
|
|
||||||
|
|
||||||
import re
|
|
||||||
|
|
||||||
from ..decorators import ensure_logged_in
|
|
||||||
from ..models import Account
|
|
||||||
from ..utils.iban import IBAN_BANKS, check_iban
|
|
||||||
from .. import decorators
|
|
||||||
from .. import db_utils
|
|
||||||
from .. import returns
|
|
||||||
|
|
||||||
bp = Blueprint('accounts', __name__, description='Bank Accounts operations')
|
|
||||||
|
|
||||||
VALID_CURRENCIES = ['RON', 'EUR', 'USD']
|
|
||||||
ACCOUNT_TYPES = ['Checking', 'Savings']
|
|
||||||
|
|
||||||
class MetaCurrenciesSchema(returns.SuccessSchema):
|
|
||||||
currencies = fields.List(fields.Str())
|
|
||||||
|
|
||||||
class MetaAccountTypesSchema(returns.SuccessSchema):
|
|
||||||
account_types = fields.List(fields.Str(), data_key='accountTypes')
|
|
||||||
|
|
||||||
class MetaValidateIbanParams(Schema):
|
|
||||||
iban = fields.Str(example='RO15RZBR0000060021338765')
|
|
||||||
|
|
||||||
class MetaValidateIbanSchema(returns.SuccessSchema):
|
|
||||||
valid = fields.Bool()
|
|
||||||
formatted_iban = fields.Str(data_key='formattedIban', optional=True)
|
|
||||||
bank_name = fields.Str(data_key='bankName', optional=True, description='Known bank for IBAN')
|
|
||||||
|
|
||||||
@bp.get('/meta/currencies')
|
|
||||||
@bp.response(200, MetaCurrenciesSchema)
|
|
||||||
def get_valid_currencies():
|
|
||||||
"""Get valid account currencies"""
|
|
||||||
return returns.success(currencies=VALID_CURRENCIES)
|
|
||||||
|
|
||||||
|
|
||||||
@bp.get('/meta/account_types')
|
|
||||||
@bp.response(200, MetaAccountTypesSchema)
|
|
||||||
def get_valid_account_types():
|
|
||||||
"""Get valid account types"""
|
|
||||||
return returns.success(account_types=ACCOUNT_TYPES)
|
|
||||||
|
|
||||||
|
|
||||||
@bp.get('/meta/validate_iban')
|
|
||||||
@bp.arguments(MetaValidateIbanParams, location='query', as_kwargs=True)
|
|
||||||
@bp.response(200, MetaValidateIbanSchema)
|
|
||||||
def get_validate_iban(iban: str):
|
|
||||||
"""Validate IBAN"""
|
|
||||||
iban = re.sub(r'\s', '', iban)
|
|
||||||
valid = len(iban) > 8 and re.match(r'^[A-Z]{2}[0-9]{2}', iban) is not None and check_iban(iban)
|
|
||||||
bank_name = None
|
|
||||||
if iban[0:2] in IBAN_BANKS:
|
|
||||||
if iban[4:8] in IBAN_BANKS[iban[0:2]]:
|
|
||||||
bank_name = IBAN_BANKS[iban[0:2]][iban[4:8]]
|
|
||||||
|
|
||||||
return returns.success(
|
|
||||||
valid=valid,
|
|
||||||
formatted_iban=re.sub(r'(.{4})', r'\1 ', iban).strip() if valid else None,
|
|
||||||
bank_name=bank_name if valid else None,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
class AccountResponseSchema(returns.SuccessSchema):
|
|
||||||
account = fields.Nested(Account.AccountSchema)
|
|
||||||
|
|
||||||
|
|
||||||
@bp.get('/<int:account_id>')
|
|
||||||
@ensure_logged_in
|
|
||||||
@bp.response(401, returns.ErrorSchema, description='Login failure or not allowed')
|
|
||||||
@bp.doc(security=[{'Token': []}])
|
|
||||||
@bp.response(200, AccountResponseSchema)
|
|
||||||
def get_account_id(account_id: int):
|
|
||||||
"""Get account by id"""
|
|
||||||
account = db_utils.get_account(account_id=account_id)
|
|
||||||
if account is None:
|
|
||||||
return returns.abort(returns.NOT_FOUND)
|
|
||||||
if decorators.user_id != db_utils.whose_account(account):
|
|
||||||
return returns.abort(returns.UNAUTHORIZED)
|
|
||||||
# account = account.to_json()
|
|
||||||
return returns.success(account=account)
|
|
||||||
|
|
||||||
|
|
||||||
@bp.get('/IBAN_<iban>')
|
|
||||||
@ensure_logged_in
|
|
||||||
@bp.response(401, returns.ErrorSchema, description='Login failure')
|
|
||||||
@bp.doc(security=[{'Token': []}])
|
|
||||||
@bp.response(200, AccountResponseSchema)
|
|
||||||
def get_account_iban(iban: str):
|
|
||||||
"""Get account by IBAN"""
|
|
||||||
account = db_utils.get_account(iban=iban)
|
|
||||||
if account is None:
|
|
||||||
return returns.abort(returns.NOT_FOUND)
|
|
||||||
if decorators.user_id != db_utils.whose_account(account):
|
|
||||||
return returns.abort(returns.UNAUTHORIZED)
|
|
||||||
# account = account.to_json()
|
|
||||||
return returns.success(account=account)
|
|
||||||
|
|
||||||
|
|
||||||
@bp.route('/')
|
|
||||||
class AccountsList(MethodView):
|
|
||||||
class CreateAccountParams(Schema):
|
|
||||||
currency = fields.String(example='RON')
|
|
||||||
account_type = fields.String(data_key='accountType', example='Checking')
|
|
||||||
custom_name = fields.String(data_key='customName', example='Daily Spending')
|
|
||||||
|
|
||||||
class CreateAccountResponseSchema(returns.SuccessSchema):
|
|
||||||
account = fields.Nested(Account.AccountSchema)
|
|
||||||
|
|
||||||
@ensure_logged_in
|
|
||||||
@bp.response(401, returns.ErrorSchema, description='Login failure')
|
|
||||||
@bp.doc(security=[{'Token': []}])
|
|
||||||
@bp.arguments(CreateAccountParams, as_kwargs=True)
|
|
||||||
@bp.response(200, CreateAccountResponseSchema)
|
|
||||||
@bp.response(422, returns.ErrorSchema, description='Invalid currency or account type')
|
|
||||||
def post(self, currency: str, account_type: str, custom_name: str):
|
|
||||||
"""Create account"""
|
|
||||||
if currency not in VALID_CURRENCIES:
|
|
||||||
return returns.abort(returns.invalid_argument('currency'))
|
|
||||||
if account_type not in ACCOUNT_TYPES:
|
|
||||||
return returns.abort(returns.invalid_argument('account_type'))
|
|
||||||
|
|
||||||
account = Account(-1, '', currency, account_type, custom_name or '')
|
|
||||||
db_utils.insert_account(decorators.user_id, account)
|
|
||||||
# return returns.success(account=account.to_json())
|
|
||||||
return returns.success(account=account)
|
|
||||||
|
|
||||||
class AccountsResponseSchema(returns.SuccessSchema):
|
|
||||||
accounts = fields.List(fields.Nested(Account.AccountSchema))
|
|
||||||
|
|
||||||
@ensure_logged_in
|
|
||||||
@bp.response(401, returns.ErrorSchema, description='Login failure')
|
|
||||||
@bp.doc(security=[{'Token': []}])
|
|
||||||
@bp.response(200, AccountsResponseSchema)
|
|
||||||
def get(self):
|
|
||||||
"""Get all accounts of user"""
|
|
||||||
return returns.success(accounts=db_utils.get_accounts(decorators.user_id))
|
|
||||||
|
|
|
@ -1,73 +0,0 @@
|
||||||
from flask.views import MethodView
|
|
||||||
from flask_smorest import Blueprint
|
|
||||||
from marshmallow import Schema, fields
|
|
||||||
from .. import returns, ram_db, decorators
|
|
||||||
from ..db_utils import get_user
|
|
||||||
from ..models import User
|
|
||||||
from ..decorators import ensure_logged_in
|
|
||||||
|
|
||||||
from pyotp import TOTP
|
|
||||||
|
|
||||||
bp = Blueprint('login', __name__, description='Login operations')
|
|
||||||
|
|
||||||
class LoginParams(Schema):
|
|
||||||
username = fields.String()
|
|
||||||
code = fields.String()
|
|
||||||
|
|
||||||
class LoginResult(returns.SuccessSchema):
|
|
||||||
token = fields.String()
|
|
||||||
|
|
||||||
class LoginSuccessSchema(returns.SuccessSchema):
|
|
||||||
token = fields.String()
|
|
||||||
|
|
||||||
@bp.route('/')
|
|
||||||
class Login(MethodView):
|
|
||||||
@bp.arguments(LoginParams, as_kwargs=True)
|
|
||||||
@bp.response(401, returns.ErrorSchema, description='Login failure')
|
|
||||||
@bp.response(200, LoginSuccessSchema)
|
|
||||||
def post(self, username: str, code: str):
|
|
||||||
"""Login via username and TOTP code"""
|
|
||||||
user: User | None = get_user(username=username)
|
|
||||||
if user is None:
|
|
||||||
return returns.abort(returns.INVALID_DETAILS)
|
|
||||||
|
|
||||||
otp = TOTP(user.otp)
|
|
||||||
if not otp.verify(code, valid_window=1):
|
|
||||||
return returns.abort(returns.INVALID_DETAILS)
|
|
||||||
|
|
||||||
token = ram_db.login_user(user.id)
|
|
||||||
return returns.success(token=token)
|
|
||||||
|
|
||||||
@ensure_logged_in
|
|
||||||
@bp.doc(security=[{'Token': []}])
|
|
||||||
@bp.response(401, returns.ErrorSchema, description='Login failure')
|
|
||||||
@bp.response(204)
|
|
||||||
def delete(self):
|
|
||||||
"""Logout"""
|
|
||||||
ram_db.logout_user(decorators.token)
|
|
||||||
|
|
||||||
@bp.post('/logout')
|
|
||||||
@ensure_logged_in
|
|
||||||
@bp.doc(security=[{'Token': []}])
|
|
||||||
@bp.response(401, returns.ErrorSchema, description='Login failure')
|
|
||||||
@bp.response(204)
|
|
||||||
def logout_route():
|
|
||||||
"""Logout"""
|
|
||||||
ram_db.logout_user(decorators.token)
|
|
||||||
|
|
||||||
@bp.route('/whoami')
|
|
||||||
class WhoAmI(MethodView):
|
|
||||||
class WhoAmISchema(returns.SuccessSchema):
|
|
||||||
user = fields.Nested(User.UserSchema)
|
|
||||||
|
|
||||||
@bp.response(401, returns.ErrorSchema, description='Login failure')
|
|
||||||
@bp.response(200, WhoAmISchema)
|
|
||||||
@bp.doc(security=[{'Token': []}])
|
|
||||||
@ensure_logged_in
|
|
||||||
def get(self):
|
|
||||||
"""Get information about currently logged in user"""
|
|
||||||
user: User | None = get_user(user_id=decorators.user_id)
|
|
||||||
# if user is not None:
|
|
||||||
# user = user.to_json()
|
|
||||||
|
|
||||||
return returns.success(user=user)
|
|
|
@ -1,60 +0,0 @@
|
||||||
from datetime import datetime
|
|
||||||
from flask.views import MethodView
|
|
||||||
from flask_smorest import Blueprint
|
|
||||||
from marshmallow import Schema, fields
|
|
||||||
|
|
||||||
from ..db_utils import get_notifications, insert_notification, mark_notification_as_read, whose_notification
|
|
||||||
from ..decorators import ensure_logged_in
|
|
||||||
from ..models import Notification
|
|
||||||
from .. import decorators, returns
|
|
||||||
|
|
||||||
bp = Blueprint('notifications', __name__, description='Notifications operations')
|
|
||||||
|
|
||||||
@bp.post('/<int:notification_id>/mark_read')
|
|
||||||
@ensure_logged_in
|
|
||||||
@bp.response(401, returns.ErrorSchema, description='Login failure or not allowed')
|
|
||||||
@bp.doc(security=[{'Token': []}])
|
|
||||||
@bp.response(201, description='Successfully marked as read')
|
|
||||||
def mark_as_read(notification_id: int):
|
|
||||||
"""Mark notification as read"""
|
|
||||||
if decorators.user_id != whose_notification(notification_id):
|
|
||||||
return returns.abort(returns.UNAUTHORIZED)
|
|
||||||
mark_notification_as_read(notification_id)
|
|
||||||
|
|
||||||
|
|
||||||
@bp.route('/')
|
|
||||||
class NotificationsList(MethodView):
|
|
||||||
class NotificationsListPostParams(Schema):
|
|
||||||
body = fields.Str(description='Text of the notification')
|
|
||||||
read = fields.Bool(default=False, description='Whether the notification was read or not')
|
|
||||||
|
|
||||||
class NotificationsListPostSchema(returns.SuccessSchema):
|
|
||||||
notification = fields.Nested(Notification.NotificationSchema)
|
|
||||||
|
|
||||||
@ensure_logged_in
|
|
||||||
@bp.response(401, returns.ErrorSchema, description='Login failure')
|
|
||||||
@bp.doc(security=[{'Token': []}])
|
|
||||||
@bp.arguments(NotificationsListPostParams, as_kwargs=True)
|
|
||||||
@bp.response(200, NotificationsListPostSchema)
|
|
||||||
def post(self, body: str, read: bool = False):
|
|
||||||
"""Post a notification to the currently logged in user
|
|
||||||
|
|
||||||
The usefulness of this endpoint is questionable besides debugging since it's a notification to self
|
|
||||||
"""
|
|
||||||
now = datetime.now()
|
|
||||||
notification = Notification.new_notification(body, now, read)
|
|
||||||
insert_notification(decorators.user_id, notification)
|
|
||||||
return returns.success(notification=notification)
|
|
||||||
|
|
||||||
class NotificationsListGetSchema(returns.SuccessSchema):
|
|
||||||
notifications = fields.List(fields.Nested(Notification.NotificationSchema))
|
|
||||||
|
|
||||||
@ensure_logged_in
|
|
||||||
@bp.response(401, returns.ErrorSchema, description='Login failure')
|
|
||||||
@bp.doc(security=[{'Token': []}])
|
|
||||||
@bp.response(200, NotificationsListGetSchema)
|
|
||||||
def get(self):
|
|
||||||
"""Get all notifications for current user"""
|
|
||||||
notifications = get_notifications(decorators.user_id)
|
|
||||||
|
|
||||||
return returns.success(notifications=notifications)
|
|
|
@ -1,122 +0,0 @@
|
||||||
from datetime import date, datetime
|
|
||||||
from flask.views import MethodView
|
|
||||||
from flask_smorest import Blueprint
|
|
||||||
from marshmallow import Schema, fields
|
|
||||||
|
|
||||||
import re
|
|
||||||
|
|
||||||
from ..decorators import ensure_logged_in
|
|
||||||
from ..db_utils import get_transactions, get_account, get_accounts, insert_transaction, whose_account, insert_notification
|
|
||||||
from ..models import Account, Notification, Transaction
|
|
||||||
from ..utils.iban import check_iban
|
|
||||||
from .. import decorators, returns
|
|
||||||
|
|
||||||
bp = Blueprint('transactions', __name__, description='Bank transfers and other transactions')
|
|
||||||
|
|
||||||
@bp.route('/')
|
|
||||||
class TransactionsList(MethodView):
|
|
||||||
class TransactionsParams(Schema):
|
|
||||||
account_id = fields.Int(min=1)
|
|
||||||
|
|
||||||
class TransactionsGetResponse(returns.SuccessSchema):
|
|
||||||
transactions = fields.List(fields.Nested(Transaction.TransactionSchema))
|
|
||||||
|
|
||||||
@ensure_logged_in
|
|
||||||
@bp.response(401, returns.ErrorSchema, description='Login failure or not allowed')
|
|
||||||
@bp.doc(security=[{'Token': []}])
|
|
||||||
@bp.arguments(TransactionsParams, as_kwargs=True, location='query')
|
|
||||||
@bp.response(200, TransactionsGetResponse)
|
|
||||||
def get(self, account_id: int):
|
|
||||||
"""Get transactions for a certain account"""
|
|
||||||
if whose_account(account_id) != decorators.user_id:
|
|
||||||
return returns.abort(returns.UNAUTHORIZED)
|
|
||||||
|
|
||||||
# return returns.success(
|
|
||||||
# transactions=[t.to_json() for t in get_transactions(account_id)]
|
|
||||||
# )
|
|
||||||
return returns.success(
|
|
||||||
transactions=get_transactions(account_id)
|
|
||||||
)
|
|
||||||
|
|
||||||
class TransactionsCreateParams(Schema):
|
|
||||||
account_id = fields.Int(min=1)
|
|
||||||
destination_iban = fields.Str()
|
|
||||||
amount = fields.Int(min=1)
|
|
||||||
description = fields.Str(default='')
|
|
||||||
|
|
||||||
class TransactionsCreateResponse(returns.SuccessSchema):
|
|
||||||
transaction = fields.Nested(Transaction.TransactionSchema)
|
|
||||||
|
|
||||||
@ensure_logged_in
|
|
||||||
@bp.response(401, returns.ErrorSchema, description='Login failure or not allowed')
|
|
||||||
@bp.response(404, returns.ErrorSchema, description='Destination account not found')
|
|
||||||
@bp.response(422, returns.ErrorSchema, description='Invalid account')
|
|
||||||
@bp.doc(security=[{'Token': []}])
|
|
||||||
@bp.arguments(TransactionsCreateParams, as_kwargs=True)
|
|
||||||
@bp.response(200, TransactionsCreateResponse)
|
|
||||||
def post(self, account_id: int, destination_iban: str, amount: int, description: str = ''):
|
|
||||||
"""Create a send_transfer transaction"""
|
|
||||||
if whose_account(account_id) != decorators.user_id:
|
|
||||||
return returns.abort(returns.UNAUTHORIZED)
|
|
||||||
|
|
||||||
account: Account = get_account(account_id=account_id)
|
|
||||||
|
|
||||||
if account is None:
|
|
||||||
return returns.abort(returns.invalid_argument('account_id'))
|
|
||||||
|
|
||||||
amount = -1 * abs(amount)
|
|
||||||
|
|
||||||
if account.balance + amount < 0:
|
|
||||||
return returns.abort(returns.NO_BALANCE)
|
|
||||||
|
|
||||||
# Check if IBAN is valid
|
|
||||||
destination_iban = re.sub(r'\s', '', destination_iban)
|
|
||||||
|
|
||||||
if not check_iban(destination_iban):
|
|
||||||
return returns.abort(returns.INVALID_IBAN)
|
|
||||||
|
|
||||||
date = datetime.now()
|
|
||||||
|
|
||||||
# Check if transaction is to another FoxBank account
|
|
||||||
reverse_transaction = None
|
|
||||||
if destination_iban[4:8] == 'FOXB':
|
|
||||||
for acc in get_accounts():
|
|
||||||
if destination_iban == acc.iban:
|
|
||||||
reverse_transaction = Transaction.new_transaction(
|
|
||||||
date_time=date,
|
|
||||||
transaction_type='receive_transfer',
|
|
||||||
status='processed',
|
|
||||||
other_party={'iban': account.iban,},
|
|
||||||
extra={
|
|
||||||
'currency': account.currency,
|
|
||||||
'amount': -amount,
|
|
||||||
'description': description,
|
|
||||||
},
|
|
||||||
)
|
|
||||||
insert_transaction(acc.id, reverse_transaction)
|
|
||||||
formatted_iban = re.sub(r'(.{4})', r'\1 ', account.iban).strip()
|
|
||||||
notification = Notification.new_notification(
|
|
||||||
body=f'Transfer of {-amount // 100}.{-amount % 100:0>2} {account.currency} received from {formatted_iban} in your {acc.custom_name or acc.account_type} account.',
|
|
||||||
date_time=date,
|
|
||||||
read=False,
|
|
||||||
)
|
|
||||||
insert_notification(acc.id, notification)
|
|
||||||
break
|
|
||||||
else:
|
|
||||||
return returns.abort(returns.NOT_FOUND)
|
|
||||||
|
|
||||||
transaction = Transaction.new_transaction(
|
|
||||||
date_time=date,
|
|
||||||
transaction_type='send_transfer',
|
|
||||||
status=('processed' if reverse_transaction is not None else 'pending'),
|
|
||||||
other_party={'iban': destination_iban,},
|
|
||||||
extra={
|
|
||||||
'currency': account.currency,
|
|
||||||
'amount': amount,
|
|
||||||
'description': description,
|
|
||||||
},
|
|
||||||
)
|
|
||||||
|
|
||||||
insert_transaction(account_id, transaction)
|
|
||||||
|
|
||||||
return returns.success(transaction=transaction)
|
|
|
@ -1,310 +0,0 @@
|
||||||
from functools import wraps
|
|
||||||
import json
|
|
||||||
import sys
|
|
||||||
from types import ModuleType
|
|
||||||
|
|
||||||
from . import db as _db
|
|
||||||
from . import models
|
|
||||||
|
|
||||||
_db_global: None | tuple[_db.get_return, int] = None
|
|
||||||
|
|
||||||
|
|
||||||
def get_db(fn):
|
|
||||||
@wraps(fn)
|
|
||||||
def wrapper(*args, **kargs):
|
|
||||||
global _db_global
|
|
||||||
if _db_global is None:
|
|
||||||
_db_global = _db.get(), 1
|
|
||||||
else:
|
|
||||||
_db_global = _db_global[0], _db_global[1] + 1
|
|
||||||
result = fn(*args, **kargs)
|
|
||||||
_db_global = _db_global[0], _db_global[1] - 1
|
|
||||||
if _db_global[1] == 0:
|
|
||||||
_db_global = None
|
|
||||||
return result
|
|
||||||
|
|
||||||
return wrapper
|
|
||||||
|
|
||||||
|
|
||||||
class Module(ModuleType):
|
|
||||||
@property
|
|
||||||
def db(self) -> _db.get_return:
|
|
||||||
if _db_global is None:
|
|
||||||
raise Exception('Function not wrapped with @get_db, db unavailable')
|
|
||||||
return _db_global[0]
|
|
||||||
|
|
||||||
|
|
||||||
@get_db
|
|
||||||
def get_user(self, username: str | None = None, user_id: int | None = None) -> models.User | None:
|
|
||||||
cur = self.db.cursor()
|
|
||||||
if username is not None:
|
|
||||||
cur.execute('select * from users where username=?', (username,))
|
|
||||||
elif user_id is not None:
|
|
||||||
cur.execute('select * from users where id=?', (user_id,))
|
|
||||||
else:
|
|
||||||
raise Exception('Neither username or user_id passed')
|
|
||||||
result = cur.fetchone()
|
|
||||||
if result is None:
|
|
||||||
return None
|
|
||||||
return models.User.from_query(result)
|
|
||||||
|
|
||||||
|
|
||||||
@get_db
|
|
||||||
def insert_user(self, user: models.User):
|
|
||||||
# Prepare user
|
|
||||||
if not user.otp:
|
|
||||||
from pyotp import random_base32
|
|
||||||
user.otp = random_base32()
|
|
||||||
|
|
||||||
cur = self.db.cursor()
|
|
||||||
cur.execute(
|
|
||||||
'insert into users(username, email, otp, fullname) values (?, ?, ?, ?)',
|
|
||||||
(user.username, user.email, user.otp, user.fullname),
|
|
||||||
)
|
|
||||||
|
|
||||||
cur.execute(
|
|
||||||
'select id from users where username = ? and email = ? and otp = ? and fullname = ?',
|
|
||||||
(user.username, user.email, user.otp, user.fullname),
|
|
||||||
)
|
|
||||||
|
|
||||||
user.id = cur.fetchone()['id']
|
|
||||||
|
|
||||||
|
|
||||||
@get_db
|
|
||||||
def get_accounts(self, user_id: int | None = None) -> list[models.Account]:
|
|
||||||
"""
|
|
||||||
Get all accounts.
|
|
||||||
If `user_id` is provided, get only the accounts for the matching user.
|
|
||||||
"""
|
|
||||||
cur = self.db.cursor()
|
|
||||||
if user_id:
|
|
||||||
cur.execute('''
|
|
||||||
select id, iban, currency, account_type, custom_name from accounts
|
|
||||||
inner join users_accounts
|
|
||||||
on accounts.id = users_accounts.account_id
|
|
||||||
where users_accounts.user_id = ?
|
|
||||||
''', (user_id,))
|
|
||||||
else:
|
|
||||||
cur.execute('select id, iban, currency, account_type, custom_name from accounts')
|
|
||||||
accounts = [models.Account.from_query(q) for q in cur.fetchall()]
|
|
||||||
|
|
||||||
for account in accounts:
|
|
||||||
cur.execute(
|
|
||||||
'select balance from V_account_balance where account_id = ?',
|
|
||||||
(account.id,),
|
|
||||||
)
|
|
||||||
|
|
||||||
result = cur.fetchone()
|
|
||||||
if result is not None:
|
|
||||||
account.balance = result['balance']
|
|
||||||
|
|
||||||
return accounts
|
|
||||||
|
|
||||||
|
|
||||||
@get_db
|
|
||||||
def get_account(self, account_id: int | None = None, iban: str | None = None) -> models.Account | None:
|
|
||||||
cur = self.db.cursor()
|
|
||||||
if account_id is not None:
|
|
||||||
cur.execute(
|
|
||||||
'select * from accounts where id=?',
|
|
||||||
(account_id,),
|
|
||||||
)
|
|
||||||
elif iban is not None:
|
|
||||||
cur.execute(
|
|
||||||
'select * from accounts where iban=?',
|
|
||||||
(iban,),
|
|
||||||
)
|
|
||||||
else:
|
|
||||||
raise Exception('Neither username or user_id passed')
|
|
||||||
result = cur.fetchone()
|
|
||||||
if result is None:
|
|
||||||
return None
|
|
||||||
account = models.Account.from_query(result)
|
|
||||||
|
|
||||||
cur.execute(
|
|
||||||
'select balance from V_account_balance where account_id = ?',
|
|
||||||
(account.id,),
|
|
||||||
)
|
|
||||||
|
|
||||||
result = cur.fetchone()
|
|
||||||
if result is not None:
|
|
||||||
account.balance = result['balance']
|
|
||||||
|
|
||||||
return account
|
|
||||||
|
|
||||||
|
|
||||||
@get_db
|
|
||||||
def whose_account(self, account: int | models.Account) -> int | None:
|
|
||||||
try:
|
|
||||||
account_id = account.id
|
|
||||||
except AttributeError:
|
|
||||||
account_id = account
|
|
||||||
|
|
||||||
cur = self.db.cursor()
|
|
||||||
cur.execute('select user_id from users_accounts where account_id = ?', (account_id,))
|
|
||||||
result = cur.fetchone()
|
|
||||||
if not result:
|
|
||||||
return None
|
|
||||||
return result['user_id']
|
|
||||||
|
|
||||||
|
|
||||||
@get_db
|
|
||||||
def insert_account(self, user_id: int, account: models.Account):
|
|
||||||
# Prepare account
|
|
||||||
ibans = [acc.iban for acc in self.get_accounts(user_id)]
|
|
||||||
if not account.iban:
|
|
||||||
from random import randint
|
|
||||||
while True:
|
|
||||||
iban = 'RO00FOXB0' + account.currency
|
|
||||||
iban += str(randint(10, 10 ** 12 - 1)).rjust(12, '0')
|
|
||||||
from .utils.iban import gen_check_digits
|
|
||||||
iban = gen_check_digits(iban)
|
|
||||||
if iban not in ibans:
|
|
||||||
break
|
|
||||||
account.iban = iban
|
|
||||||
|
|
||||||
cur = self.db.cursor()
|
|
||||||
cur.execute(
|
|
||||||
'insert into accounts(iban, currency, account_type, custom_name) values (?, ?, ?, ?)',
|
|
||||||
(account.iban, account.currency, account.account_type, account.custom_name),
|
|
||||||
)
|
|
||||||
|
|
||||||
cur.execute(
|
|
||||||
'select id from accounts where iban = ?',
|
|
||||||
(account.iban,),
|
|
||||||
)
|
|
||||||
account.id = cur.fetchone()['id']
|
|
||||||
|
|
||||||
cur.execute(
|
|
||||||
'insert into users_accounts(user_id, account_id) VALUES (?, ?)',
|
|
||||||
(user_id, account.id)
|
|
||||||
)
|
|
||||||
|
|
||||||
self.db.commit()
|
|
||||||
|
|
||||||
@get_db
|
|
||||||
def get_transactions(self, account_id: int) -> list[models.Transaction]:
|
|
||||||
cur = self.db.cursor()
|
|
||||||
cur.execute(
|
|
||||||
'select transaction_id from accounts_transactions where account_id = ?',
|
|
||||||
(account_id,),
|
|
||||||
)
|
|
||||||
|
|
||||||
transactions = []
|
|
||||||
for tid in (row['transaction_id'] for row in cur.fetchall()):
|
|
||||||
cur.execute(
|
|
||||||
'select * from transactions where id = ?',
|
|
||||||
(tid,),
|
|
||||||
)
|
|
||||||
|
|
||||||
db_res = cur.fetchone()
|
|
||||||
if db_res is None:
|
|
||||||
continue
|
|
||||||
transactions.append(models.Transaction.from_query(db_res))
|
|
||||||
|
|
||||||
return transactions
|
|
||||||
|
|
||||||
@get_db
|
|
||||||
def insert_transaction(self, account_id: int, transaction: models.Transaction):
|
|
||||||
cur = self.db.cursor()
|
|
||||||
cur.execute(
|
|
||||||
'insert into transactions(datetime, other_party, status, type, extra) values (?, ?, ?, ?, ?)',
|
|
||||||
(
|
|
||||||
transaction.date_time.isoformat(),
|
|
||||||
json.dumps(transaction.other_party),
|
|
||||||
transaction.status,
|
|
||||||
transaction.transaction_type,
|
|
||||||
json.dumps(transaction.extra),
|
|
||||||
),
|
|
||||||
)
|
|
||||||
|
|
||||||
cur.execute(
|
|
||||||
'select id from transactions where datetime = ? and other_party = ? and status = ? and type = ? and extra = ?',
|
|
||||||
(
|
|
||||||
transaction.date_time.isoformat(),
|
|
||||||
json.dumps(transaction.other_party),
|
|
||||||
transaction.status,
|
|
||||||
transaction.transaction_type,
|
|
||||||
json.dumps(transaction.extra),
|
|
||||||
),
|
|
||||||
)
|
|
||||||
transaction.id = cur.fetchone()['id']
|
|
||||||
|
|
||||||
cur.execute(
|
|
||||||
'insert into accounts_transactions(account_id, transaction_id) values (?, ?)',
|
|
||||||
(account_id, transaction.id),
|
|
||||||
)
|
|
||||||
|
|
||||||
self.db.commit()
|
|
||||||
|
|
||||||
@get_db
|
|
||||||
def get_notifications(self, user_id: int) -> list[models.Notification]:
|
|
||||||
cur = self.db.cursor()
|
|
||||||
|
|
||||||
cur.execute(
|
|
||||||
'''
|
|
||||||
select n.id, n.body, n.datetime, n.read
|
|
||||||
from notifications as n
|
|
||||||
inner join users_notifications on n.id = users_notifications.notification_id
|
|
||||||
where users_notifications.user_id = ?
|
|
||||||
''',
|
|
||||||
(user_id,),
|
|
||||||
)
|
|
||||||
|
|
||||||
return [models.Notification.from_query(q) for q in cur.fetchall()]
|
|
||||||
|
|
||||||
@get_db
|
|
||||||
def insert_notification(self, user_id: int, notification: models.Notification):
|
|
||||||
cur = self.db.cursor()
|
|
||||||
|
|
||||||
cur.execute(
|
|
||||||
'insert into notifications(body, datetime, read) values (?, ?, ?)',
|
|
||||||
(
|
|
||||||
notification.body,
|
|
||||||
notification.date_time.isoformat(),
|
|
||||||
1 if notification.read else 0,
|
|
||||||
),
|
|
||||||
)
|
|
||||||
|
|
||||||
cur.execute(
|
|
||||||
'select id from notifications where body = ? and datetime = ? and read = ?',
|
|
||||||
(
|
|
||||||
notification.body,
|
|
||||||
notification.date_time.isoformat(),
|
|
||||||
1 if notification.read else 0,
|
|
||||||
),
|
|
||||||
)
|
|
||||||
notification.id = cur.fetchone()['id']
|
|
||||||
|
|
||||||
cur.execute(
|
|
||||||
'insert into users_notifications values (?, ?)',
|
|
||||||
(user_id, notification.id,),
|
|
||||||
)
|
|
||||||
|
|
||||||
self.db.commit()
|
|
||||||
|
|
||||||
@get_db
|
|
||||||
def whose_notification(self, notification: int | models.Notification) -> int | None:
|
|
||||||
try:
|
|
||||||
notification_id = notification.id
|
|
||||||
except AttributeError:
|
|
||||||
notification_id = notification
|
|
||||||
|
|
||||||
cur = self.db.cursor()
|
|
||||||
cur.execute('select user_id from users_notifications where notification_id = ?', (notification_id,))
|
|
||||||
result = cur.fetchone()
|
|
||||||
if not result:
|
|
||||||
return None
|
|
||||||
return result[0]
|
|
||||||
|
|
||||||
@get_db
|
|
||||||
def mark_notification_as_read(self, notification_id: int):
|
|
||||||
cur = self.db.cursor()
|
|
||||||
cur.execute(
|
|
||||||
'update notifications set read = 1 where id = ?',
|
|
||||||
(notification_id,),
|
|
||||||
)
|
|
||||||
self.db.commit()
|
|
||||||
|
|
||||||
|
|
||||||
sys.modules[__name__] = Module(__name__)
|
|
|
@ -1,74 +0,0 @@
|
||||||
import sys
|
|
||||||
from types import ModuleType
|
|
||||||
from flask import request
|
|
||||||
from http import HTTPStatus
|
|
||||||
from functools import wraps
|
|
||||||
from . import ram_db
|
|
||||||
from . import returns
|
|
||||||
|
|
||||||
_token: str | None = None
|
|
||||||
_user_id: int | None = None
|
|
||||||
|
|
||||||
class Module(ModuleType):
|
|
||||||
def no_content(self, fn):
|
|
||||||
"""
|
|
||||||
Allows a Flask route to return None, which is converted into
|
|
||||||
HTTP 201 No Content.
|
|
||||||
"""
|
|
||||||
@wraps(fn)
|
|
||||||
def wrapper(*args, **kargs):
|
|
||||||
result = fn(*args, **kargs)
|
|
||||||
if result is None:
|
|
||||||
return None, HTTPStatus.NO_CONTENT
|
|
||||||
else:
|
|
||||||
return result
|
|
||||||
return wrapper
|
|
||||||
|
|
||||||
@property
|
|
||||||
def token(self) -> str:
|
|
||||||
if _token is None:
|
|
||||||
raise Exception('No token available')
|
|
||||||
return _token
|
|
||||||
|
|
||||||
@property
|
|
||||||
def user_id(self) -> int:
|
|
||||||
if _user_id is None:
|
|
||||||
raise Exception('No user_id available')
|
|
||||||
return _user_id
|
|
||||||
|
|
||||||
def ensure_logged_in(self, fn):
|
|
||||||
"""
|
|
||||||
Ensure the user is logged in by providing an Authorization: Bearer token
|
|
||||||
header.
|
|
||||||
|
|
||||||
@param token whether the token should be supplied after validation
|
|
||||||
@param user_id whether the user_id should be supplied after validation
|
|
||||||
@return decorator which supplies the requested parameters
|
|
||||||
"""
|
|
||||||
@wraps(fn)
|
|
||||||
def wrapper(*args, **kargs):
|
|
||||||
token = request.headers.get('Authorization', None)
|
|
||||||
if token is None:
|
|
||||||
return returns.abort(returns.NO_AUTHORIZATION)
|
|
||||||
if not token.startswith('Bearer '):
|
|
||||||
return returns.abort(returns.INVALID_AUTHORIZATION)
|
|
||||||
token = token[7:]
|
|
||||||
user_id = ram_db.get_user(token)
|
|
||||||
if user_id is None:
|
|
||||||
return returns.abort(returns.INVALID_AUTHORIZATION)
|
|
||||||
|
|
||||||
global _token
|
|
||||||
_token = token
|
|
||||||
global _user_id
|
|
||||||
_user_id = user_id
|
|
||||||
|
|
||||||
result = fn(*args, **kargs)
|
|
||||||
|
|
||||||
_token = None
|
|
||||||
_user_id = None
|
|
||||||
|
|
||||||
return result
|
|
||||||
|
|
||||||
return wrapper
|
|
||||||
|
|
||||||
sys.modules[__name__] = Module(__name__)
|
|
|
@ -1,175 +0,0 @@
|
||||||
from dataclasses import dataclass, field
|
|
||||||
from marshmallow import Schema, fields
|
|
||||||
from datetime import datetime
|
|
||||||
|
|
||||||
@dataclass
|
|
||||||
class User:
|
|
||||||
id: int
|
|
||||||
username: str
|
|
||||||
email: str
|
|
||||||
otp: str
|
|
||||||
fullname: str
|
|
||||||
|
|
||||||
class UserSchema(Schema):
|
|
||||||
id = fields.Int(required=False)
|
|
||||||
username = fields.String()
|
|
||||||
email = fields.String()
|
|
||||||
otp = fields.String(load_only=True, required=False)
|
|
||||||
fullname = fields.String()
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def new_user(username: str, email: str, fullname: str) -> 'User':
|
|
||||||
return User(
|
|
||||||
id=-1,
|
|
||||||
username=username,
|
|
||||||
email=email,
|
|
||||||
otp='',
|
|
||||||
fullname=fullname,
|
|
||||||
)
|
|
||||||
|
|
||||||
def to_json(self, include_otp=False, include_id=False):
|
|
||||||
result = {
|
|
||||||
'username': self.username,
|
|
||||||
'email': self.email,
|
|
||||||
'fullname': self.fullname,
|
|
||||||
}
|
|
||||||
if include_id:
|
|
||||||
result['id'] = self.id
|
|
||||||
if include_otp:
|
|
||||||
result['otp'] = self.otp
|
|
||||||
return result
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def from_query(cls, query_result):
|
|
||||||
return cls(*query_result)
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
|
||||||
class Account:
|
|
||||||
id: int
|
|
||||||
iban: str
|
|
||||||
currency: str
|
|
||||||
account_type: str
|
|
||||||
custom_name: str
|
|
||||||
balance: int = field(default=0)
|
|
||||||
|
|
||||||
class AccountSchema(Schema):
|
|
||||||
id = fields.Int(required=False)
|
|
||||||
iban = fields.Str()
|
|
||||||
currency = fields.Str()
|
|
||||||
account_type = fields.Str(data_key='accountType')
|
|
||||||
custom_name = fields.Str(data_key='customName')
|
|
||||||
balance = fields.Int()
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def new_account(currency: str, account_type: str, custom_name: str = '') -> 'Account':
|
|
||||||
return Account(
|
|
||||||
id=-1,
|
|
||||||
iban='',
|
|
||||||
currency=currency,
|
|
||||||
account_type=account_type,
|
|
||||||
custom_name=custom_name,
|
|
||||||
)
|
|
||||||
|
|
||||||
def to_json(self, include_id=True):
|
|
||||||
result = {
|
|
||||||
'iban': self.iban,
|
|
||||||
'currency': self.currency,
|
|
||||||
'accountType': self.account_type,
|
|
||||||
'customName': self.custom_name,
|
|
||||||
'balance': self.balance,
|
|
||||||
}
|
|
||||||
if include_id:
|
|
||||||
result['id'] = self.id
|
|
||||||
return result
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def from_query(cls, query_result):
|
|
||||||
return cls(*query_result)
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
|
||||||
class Transaction:
|
|
||||||
id: int
|
|
||||||
date_time: datetime
|
|
||||||
other_party: str
|
|
||||||
status: str
|
|
||||||
transaction_type: str
|
|
||||||
extra: str
|
|
||||||
|
|
||||||
class TransactionSchema(Schema):
|
|
||||||
id = fields.Int(required=False)
|
|
||||||
date_time = fields.DateTime(data_key='datetime')
|
|
||||||
other_party = fields.Dict(keys=fields.Str(), values=fields.Raw(), data_key='otherParty')
|
|
||||||
status = fields.Str()
|
|
||||||
transaction_type = fields.Str(data_key='transactionType')
|
|
||||||
extra = fields.Dict(keys=fields.Str(), values=fields.Raw())
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def new_transaction(date_time: datetime, other_party: str, status: str, transaction_type: str, extra: str = '') -> 'Transaction':
|
|
||||||
return Transaction(
|
|
||||||
id=-1,
|
|
||||||
date_time=date_time,
|
|
||||||
other_party=other_party,
|
|
||||||
status=status,
|
|
||||||
transaction_type=transaction_type,
|
|
||||||
extra=extra,
|
|
||||||
)
|
|
||||||
|
|
||||||
def to_json(self, include_id=True):
|
|
||||||
result = {
|
|
||||||
'datetime': self.date_time.isoformat(),
|
|
||||||
'otherParty': self.other_party,
|
|
||||||
'status': self.status,
|
|
||||||
'transactionType': self.transaction_type,
|
|
||||||
'extra': self.extra,
|
|
||||||
}
|
|
||||||
if include_id:
|
|
||||||
result['id'] = self.id
|
|
||||||
return result
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def from_query(cls, query_result):
|
|
||||||
import json
|
|
||||||
|
|
||||||
query_result = list(query_result)
|
|
||||||
if type(query_result[1]) is str:
|
|
||||||
query_result[1] = datetime.fromisoformat(query_result[1])
|
|
||||||
if type(query_result[2]) is str:
|
|
||||||
query_result[2] = json.loads(query_result[2])
|
|
||||||
if type(query_result[5]) is str:
|
|
||||||
query_result[5] = json.loads(query_result[5])
|
|
||||||
|
|
||||||
return cls(*query_result)
|
|
||||||
|
|
||||||
@dataclass
|
|
||||||
class Notification:
|
|
||||||
id: int
|
|
||||||
body: str
|
|
||||||
date_time: datetime
|
|
||||||
read: bool
|
|
||||||
|
|
||||||
class NotificationSchema(Schema):
|
|
||||||
id = fields.Int(required=False)
|
|
||||||
body = fields.Str()
|
|
||||||
date_time = fields.DateTime(data_key='datetime')
|
|
||||||
read = fields.Bool()
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def new_notification(body: str, date_time: datetime, read: bool = False) -> 'Notification':
|
|
||||||
return Notification(
|
|
||||||
id=-1,
|
|
||||||
body=body,
|
|
||||||
date_time=date_time,
|
|
||||||
read=read,
|
|
||||||
)
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def from_query(cls, query_result):
|
|
||||||
query_result = list(query_result)
|
|
||||||
if type(query_result[2]) is str:
|
|
||||||
query_result[2] = datetime.fromisoformat(query_result[2])
|
|
||||||
if type(query_result[3]) is not bool:
|
|
||||||
query_result[3] = bool(query_result[3])
|
|
||||||
|
|
||||||
return cls(*query_result)
|
|
|
@ -1,108 +0,0 @@
|
||||||
from http import HTTPStatus as _HTTPStatus
|
|
||||||
from typing import Any
|
|
||||||
|
|
||||||
|
|
||||||
def _make_error(http_status, code: str, message: str | None = None):
|
|
||||||
try:
|
|
||||||
http_status = http_status[0]
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
|
|
||||||
payload = {
|
|
||||||
'status': 'error',
|
|
||||||
'code': code,
|
|
||||||
}
|
|
||||||
|
|
||||||
if message is not None:
|
|
||||||
payload['message'] = message
|
|
||||||
|
|
||||||
return payload, http_status
|
|
||||||
|
|
||||||
|
|
||||||
# General
|
|
||||||
|
|
||||||
INVALID_REQUEST = _make_error(
|
|
||||||
_HTTPStatus.BAD_REQUEST,
|
|
||||||
'general/invalid_request',
|
|
||||||
)
|
|
||||||
|
|
||||||
NOT_FOUND = _make_error(
|
|
||||||
_HTTPStatus.NOT_FOUND,
|
|
||||||
'general/not_found',
|
|
||||||
)
|
|
||||||
|
|
||||||
def invalid_argument(argname: str) -> tuple[Any, int]:
|
|
||||||
return _make_error(
|
|
||||||
_HTTPStatus.UNPROCESSABLE_ENTITY,
|
|
||||||
'general/invalid_argument',
|
|
||||||
message=f'Invalid argument: {argname}',
|
|
||||||
)
|
|
||||||
|
|
||||||
# Login
|
|
||||||
|
|
||||||
INVALID_DETAILS = _make_error(
|
|
||||||
_HTTPStatus.UNAUTHORIZED,
|
|
||||||
'login/invalid_details',
|
|
||||||
)
|
|
||||||
|
|
||||||
NO_AUTHORIZATION = _make_error(
|
|
||||||
_HTTPStatus.UNAUTHORIZED,
|
|
||||||
'login/no_authorization',
|
|
||||||
)
|
|
||||||
|
|
||||||
INVALID_AUTHORIZATION = _make_error(
|
|
||||||
_HTTPStatus.UNAUTHORIZED,
|
|
||||||
'login/invalid_authorization',
|
|
||||||
)
|
|
||||||
|
|
||||||
UNAUTHORIZED = _make_error(
|
|
||||||
_HTTPStatus.UNAUTHORIZED,
|
|
||||||
'login/unauthorized',
|
|
||||||
"You are logged in but the resource you're trying to access isn't available to you",
|
|
||||||
)
|
|
||||||
|
|
||||||
# Transactions
|
|
||||||
|
|
||||||
NO_BALANCE = _make_error(
|
|
||||||
_HTTPStatus.BAD_REQUEST,
|
|
||||||
'transaction/no_balance',
|
|
||||||
'Not enough balance to make the transaction',
|
|
||||||
)
|
|
||||||
|
|
||||||
INVALID_IBAN = _make_error(
|
|
||||||
_HTTPStatus.BAD_REQUEST,
|
|
||||||
'transaction/invalid_iban',
|
|
||||||
'Recipient IBAN is invalid',
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
# Success
|
|
||||||
|
|
||||||
def success(http_status: Any = _HTTPStatus.OK, /, **kargs):
|
|
||||||
try:
|
|
||||||
http_status = http_status[0]
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
|
|
||||||
return dict(kargs, status='success'), http_status
|
|
||||||
|
|
||||||
# Schemas
|
|
||||||
|
|
||||||
from marshmallow import Schema, fields, validate
|
|
||||||
|
|
||||||
class ErrorSchema(Schema):
|
|
||||||
status = fields.Str(default='error', validate=validate.Equal('error'))
|
|
||||||
code = fields.Str()
|
|
||||||
message = fields.Str(required=False)
|
|
||||||
|
|
||||||
class SuccessSchema(Schema):
|
|
||||||
status = fields.Str(default='success', validate=validate.Equal('success'))
|
|
||||||
|
|
||||||
# smorest
|
|
||||||
|
|
||||||
def abort(result: tuple[Any, int]):
|
|
||||||
try:
|
|
||||||
from flask_smorest import abort as _abort
|
|
||||||
_abort(result[1], response=result)
|
|
||||||
except ImportError:
|
|
||||||
return result
|
|
|
@ -1,55 +0,0 @@
|
||||||
from .string import str_range_replace
|
|
||||||
|
|
||||||
|
|
||||||
IBAN_BANKS = {
|
|
||||||
'RO': {
|
|
||||||
'NBOR': 'BANCA NATIONALA A ROMANIEI',
|
|
||||||
'BUCU': 'ALPHA BANK ROMANIA SA',
|
|
||||||
'CARP': 'BANCA COMERCIALA CARPATICA SA',
|
|
||||||
'RNCB': 'BANCA COMERCIALA ROMANA SA',
|
|
||||||
'BRDE': 'BANCA ROMANA PENTRU DEZVOLTARE',
|
|
||||||
'BRMA': 'BANCA ROMANEASCA SA',
|
|
||||||
'BTRL': 'BANCA TRANSILVANIA SA',
|
|
||||||
'DAFB': 'BANK LEUMI ROMANIA SA',
|
|
||||||
'CECE': 'CASA DE ECONOMII SI CONSEMNATIUNI CEC SA',
|
|
||||||
'CITI': 'CITIBANK ROMANIA SA',
|
|
||||||
'UGBI': 'GARANTIBANK INTERNATIONAL NV - SUCURSALA ROMANIA',
|
|
||||||
'INGB': 'ING BANK NV',
|
|
||||||
'BREL': 'LIBRA BANK SA',
|
|
||||||
'BNRB': 'OTP BANK ROMANIA SA',
|
|
||||||
'RZBR': 'RAIFFEISEN BANK SA',
|
|
||||||
'TREZ': 'TREZORERIA STATULUI',
|
|
||||||
'BACX': 'UNICREDIT BANK SA',
|
|
||||||
'FOXB': 'FOXBANK',
|
|
||||||
},
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
def c_to_iban_i(c: str) -> int:
|
|
||||||
a = ord(c)
|
|
||||||
if a in range(48, 58):
|
|
||||||
return a - 48
|
|
||||||
elif a in range(65, 91):
|
|
||||||
return a - 65 + 10
|
|
||||||
elif a in range(97, 123):
|
|
||||||
return a - 97 + 10
|
|
||||||
else:
|
|
||||||
raise ValueError(f'Invalid IBAN character: {c} (ord: {a})')
|
|
||||||
|
|
||||||
|
|
||||||
def iban_to_int(iban: str) -> int:
|
|
||||||
iban = iban[4:] + iban[0:4]
|
|
||||||
return int(''.join(map(str, map(c_to_iban_i, iban))))
|
|
||||||
|
|
||||||
|
|
||||||
def check_iban(iban: str) -> bool:
|
|
||||||
num = iban_to_int(iban)
|
|
||||||
return num % 97 == 1
|
|
||||||
|
|
||||||
|
|
||||||
def gen_check_digits(iban: str) -> str:
|
|
||||||
iban = str_range_replace(iban, '00', 2, 4)
|
|
||||||
num = iban_to_int(iban)
|
|
||||||
check = 98 - (num % 97)
|
|
||||||
iban = str_range_replace(iban, str(check).rjust(2, '0'), 2, 4)
|
|
||||||
return iban
|
|
|
@ -1,7 +0,0 @@
|
||||||
def str_range_replace(
|
|
||||||
input: str,
|
|
||||||
replace_with: str,
|
|
||||||
range_start: int | None = None,
|
|
||||||
range_end: int | None = None,
|
|
||||||
) -> str:
|
|
||||||
return input[:range_start] + replace_with + input[range_end:]
|
|
|
@ -58,12 +58,4 @@ create table users_notifications (
|
||||||
notification_id integer UNIQUE not null,
|
notification_id integer UNIQUE not null,
|
||||||
foreign key (user_id) references users (id),
|
foreign key (user_id) references users (id),
|
||||||
foreign key (notification_id) references notifications (id)
|
foreign key (notification_id) references notifications (id)
|
||||||
);
|
);
|
||||||
|
|
||||||
create view V_account_balance as
|
|
||||||
select
|
|
||||||
accounts_transactions.account_id as "account_id",
|
|
||||||
sum(json_extract(transactions.extra, '$.amount')) as "balance"
|
|
||||||
from transactions
|
|
||||||
inner join accounts_transactions on accounts_transactions.transaction_id = transactions.id
|
|
||||||
group by accounts_transactions.account_id;
|
|
73
server/login.py
Normal file
73
server/login.py
Normal file
|
@ -0,0 +1,73 @@
|
||||||
|
from functools import wraps
|
||||||
|
from flask import Blueprint, request
|
||||||
|
|
||||||
|
from pyotp import TOTP
|
||||||
|
|
||||||
|
import db_utils
|
||||||
|
from decorators import no_content
|
||||||
|
import models
|
||||||
|
import ram_db
|
||||||
|
import returns
|
||||||
|
|
||||||
|
login = Blueprint('login', __name__)
|
||||||
|
|
||||||
|
@login.post('/')
|
||||||
|
def make_login():
|
||||||
|
try:
|
||||||
|
username = request.json['username']
|
||||||
|
code = request.json['code']
|
||||||
|
except (TypeError, KeyError):
|
||||||
|
return returns.INVALID_REQUEST
|
||||||
|
|
||||||
|
user: models.User | None = db_utils.get_user(username=username)
|
||||||
|
if user is None:
|
||||||
|
return returns.INVALID_DETAILS
|
||||||
|
|
||||||
|
otp = TOTP(user.otp)
|
||||||
|
if not otp.verify(code, valid_window=1):
|
||||||
|
return returns.INVALID_DETAILS
|
||||||
|
|
||||||
|
token = ram_db.login_user(user.id)
|
||||||
|
return returns.success(token=token)
|
||||||
|
|
||||||
|
def ensure_logged_in(token=False, user_id=False):
|
||||||
|
def decorator(fn):
|
||||||
|
pass_token = token
|
||||||
|
pass_user_id = user_id
|
||||||
|
@wraps(fn)
|
||||||
|
def wrapper(*args, **kargs):
|
||||||
|
token = request.headers.get('Authorization', None)
|
||||||
|
if token is None:
|
||||||
|
return returns.NO_AUTHORIZATION
|
||||||
|
if not token.startswith('Bearer '):
|
||||||
|
return returns.INVALID_AUTHORIZATION
|
||||||
|
token = token[7:]
|
||||||
|
user_id = ram_db.get_user(token)
|
||||||
|
if user_id is None:
|
||||||
|
return returns.INVALID_AUTHORIZATION
|
||||||
|
|
||||||
|
if pass_user_id and pass_token:
|
||||||
|
return fn(user_id=user_id, token=token, *args, **kargs)
|
||||||
|
elif pass_user_id:
|
||||||
|
return fn(user_id=user_id, *args, **kargs)
|
||||||
|
elif pass_token:
|
||||||
|
return fn(token=token, *args, **kargs)
|
||||||
|
else:
|
||||||
|
return fn(*args, **kargs)
|
||||||
|
return wrapper
|
||||||
|
return decorator
|
||||||
|
|
||||||
|
@login.post('/logout')
|
||||||
|
@ensure_logged_in(token=True)
|
||||||
|
@no_content
|
||||||
|
def logout(token: str):
|
||||||
|
ram_db.logout_user(token)
|
||||||
|
|
||||||
|
@login.get('/whoami')
|
||||||
|
@ensure_logged_in(user_id=True)
|
||||||
|
def whoami(user_id: int):
|
||||||
|
user: models.User | None = db_utils.get_user(user_id=user_id)
|
||||||
|
if user is not None:
|
||||||
|
user = user.to_json()
|
||||||
|
|
||||||
|
return returns.success(user=user)
|
25
server/models.py
Normal file
25
server/models.py
Normal file
|
@ -0,0 +1,25 @@
|
||||||
|
from dataclasses import dataclass
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class User:
|
||||||
|
id: int
|
||||||
|
username: str
|
||||||
|
email: str
|
||||||
|
otp: str
|
||||||
|
fullname: str
|
||||||
|
|
||||||
|
def to_json(self, include_otp=False, include_id=False):
|
||||||
|
result = {
|
||||||
|
'username': self.username,
|
||||||
|
'email': self.email,
|
||||||
|
'fullname': self.fullname,
|
||||||
|
}
|
||||||
|
if include_id:
|
||||||
|
result['id'] = self.id
|
||||||
|
if include_otp:
|
||||||
|
result['otp'] = self.otp
|
||||||
|
return result
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def from_query(cls, query_result):
|
||||||
|
return cls(*query_result)
|
|
@ -1,4 +1,5 @@
|
||||||
from datetime import datetime, timedelta
|
from datetime import datetime, timedelta
|
||||||
|
from types import TracebackType
|
||||||
from uuid import uuid4
|
from uuid import uuid4
|
||||||
|
|
||||||
USED_TOKENS = set()
|
USED_TOKENS = set()
|
46
server/returns.py
Normal file
46
server/returns.py
Normal file
|
@ -0,0 +1,46 @@
|
||||||
|
from http import HTTPStatus as _HTTPStatus
|
||||||
|
|
||||||
|
def _make_error(http_status, code: str):
|
||||||
|
try:
|
||||||
|
http_status = http_status[0]
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
return {
|
||||||
|
'status': 'error',
|
||||||
|
'code': code,
|
||||||
|
}, http_status
|
||||||
|
|
||||||
|
# General
|
||||||
|
|
||||||
|
INVALID_REQUEST = _make_error(
|
||||||
|
_HTTPStatus.BAD_REQUEST,
|
||||||
|
'general/invalid_request',
|
||||||
|
)
|
||||||
|
|
||||||
|
# Login
|
||||||
|
|
||||||
|
INVALID_DETAILS = _make_error(
|
||||||
|
_HTTPStatus.UNAUTHORIZED,
|
||||||
|
'login/invalid_details',
|
||||||
|
)
|
||||||
|
|
||||||
|
NO_AUTHORIZATION = _make_error(
|
||||||
|
_HTTPStatus.UNAUTHORIZED,
|
||||||
|
'login/no_authorization',
|
||||||
|
)
|
||||||
|
|
||||||
|
INVALID_AUTHORIZATION = _make_error(
|
||||||
|
_HTTPStatus.UNAUTHORIZED,
|
||||||
|
'login/invalid_authorization',
|
||||||
|
)
|
||||||
|
|
||||||
|
# Success
|
||||||
|
|
||||||
|
def success(http_status=_HTTPStatus.OK, /, **kargs):
|
||||||
|
try:
|
||||||
|
http_status = http_status[0]
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
return dict(kargs, status='success'), http_status
|
22
server/server.py
Executable file → Normal file
22
server/server.py
Executable file → Normal file
|
@ -1,18 +1,14 @@
|
||||||
#! /usr/bin/env python3
|
from flask import Flask
|
||||||
from foxbank_server import create_app
|
from flask_cors import CORS
|
||||||
|
|
||||||
app = create_app()
|
import db
|
||||||
# api = Api(app)
|
|
||||||
# CORS(app)
|
|
||||||
# db.init_app(app)
|
|
||||||
|
|
||||||
# from login import login
|
app = Flask(__name__)
|
||||||
# app.register_blueprint(login, url_prefix='/login')
|
CORS(app)
|
||||||
|
db.init_app(app)
|
||||||
|
|
||||||
# from bank_accounts import blueprint as ba_bp, namespace as ba_ns
|
from login import login
|
||||||
# app.register_blueprint(ba_bp, url_prefix='/accounts')
|
app.register_blueprint(login, url_prefix='/login')
|
||||||
|
|
||||||
# accounts_ns = api.add_namespace(ba_ns, '/accounts')
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
if __name__ == '__main__':
|
||||||
app.run(debug=True)
|
app.run()
|
||||||
|
|
|
@ -1,2 +0,0 @@
|
||||||
[pycodestyle]
|
|
||||||
ignore = E402
|
|
Loading…
Add table
Reference in a new issue