88 lines
2.8 KiB
Python
88 lines
2.8 KiB
Python
from dotenv import load_dotenv
|
|
from flask import Flask, jsonify
|
|
from flask_jwt_extended import JWTManager
|
|
from jwt import ExpiredSignatureError
|
|
from models import db, RevokedToken
|
|
import os
|
|
from sqlalchemy import text
|
|
from sqlalchemy.exc import DatabaseError
|
|
import time
|
|
from utils import init_db
|
|
from views import user_bp
|
|
from werkzeug.exceptions import HTTPException
|
|
|
|
MAX_RETRIES = 100
|
|
|
|
def wait_for_db():
|
|
for retries in range(MAX_RETRIES):
|
|
try:
|
|
with db.engine.connect() as connection:
|
|
connection.execute(text("SELECT 1"))
|
|
print("Successfully connected with database.")
|
|
return
|
|
except DatabaseError:
|
|
print(f"Waiting for database... (retry {retries + 1})")
|
|
time.sleep(3)
|
|
print("Failed to connect to database.")
|
|
raise Exception("Database not ready after multiple retries.")
|
|
|
|
|
|
def create_app(config_name="default"):
|
|
"""Creates and returns a new instance of Flask app."""
|
|
load_dotenv()
|
|
app = Flask(__name__)
|
|
|
|
# Database settings
|
|
if config_name == "testing":
|
|
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:" # Database in memory
|
|
app.config["TESTING"] = True
|
|
else:
|
|
app.config["SQLALCHEMY_DATABASE_URI"] = os.getenv("SQLALCHEMY_DATABASE_URI")
|
|
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
|
|
|
|
# JWT settings
|
|
app.config["JWT_SECRET_KEY"] = os.getenv("JWT_SECRET_KEY", "changeme")
|
|
|
|
# Blueprints registration
|
|
app.register_blueprint(user_bp)
|
|
|
|
# Database and JWT initialization
|
|
db.init_app(app)
|
|
jwt = JWTManager(app)
|
|
|
|
# Function to check if JWT token is revoked
|
|
@jwt.token_in_blocklist_loader
|
|
def check_if_token_revoked(jwt_header, jwt_payload):
|
|
token = db.session.get(RevokedToken, jwt_payload["jti"])
|
|
return token is not None
|
|
|
|
# Global error handler
|
|
@app.errorhandler(Exception)
|
|
def global_error_handler(error):
|
|
if isinstance(error, HTTPException):
|
|
response = jsonify({"error": error.description})
|
|
response.status_code = error.code
|
|
elif isinstance(error, ExpiredSignatureError):
|
|
response = jsonify({"error": "Token has expired"})
|
|
response.status_code = 401
|
|
else: # All other errors
|
|
response = jsonify({"error": str(error)})
|
|
response.status_code = 500
|
|
return response
|
|
|
|
# Fill database by initial values (only if we are not testing)
|
|
with app.app_context():
|
|
wait_for_db()
|
|
db.create_all()
|
|
if config_name != "testing":
|
|
init_db()
|
|
return app
|
|
|
|
|
|
# Server start only if we run app directly
|
|
if __name__ == "__main__":
|
|
from waitress import serve
|
|
app = create_app()
|
|
port = os.getenv("APP_PORT", "80")
|
|
serve(app, host="0.0.0.0", port=port)
|