12592571620

Committed 03 Jan 2025 04:04AM UTC coverage: 94.427% (-0.4%) from 94.865%

Build # 12592571620

Build Type

Pull #157

github

Committed by

SeaweedbrainCY

Commit Message

Add build info

Pull Request Pull Request #157: Add healthcheck endpoint

Run Details

18 of 46 new or added lines in 3 files covered. (39.13%)

110 existing lines in 3 files now uncovered.

11081 of 11735 relevant lines covered (94.43%)

0.94 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

54.55

api/app.py

import connexion
from flask_cors import CORS
from environment import conf
from database.db import db
from zero_totp_db_model.model_init import init_db
import uvicorn
from asgiref.wsgi import WsgiToAsgi
from starlette.middleware.cors import CORSMiddleware
from connexion.middleware import MiddlewarePosition
from environment import logging
import contextlib
from flask_apscheduler import APScheduler
from monitoring.sentry import sentry_configuration
from flask_migrate import Migrate
import datetime as dt
from flask import request, redirect, make_response


def create_app():
    app_instance = connexion.FlaskApp(__name__, specification_dir="./openAPI/")
    app_instance.add_api("swagger.yml")

    app = app_instance.app

    app.config["SQLALCHEMY_DATABASE_URI"] = conf.database.database_uri
    app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
    app.config["PROPAGATE_EXCEPTIONS"] = True
    app.secret_key = conf.api.flask_secret_key
    

    
    db.init_app(app)
    sentry_configuration() #optional
    init_db(db)
    
    
    

    return app_instance, app
app, flask = create_app()
migrate = Migrate(flask, db)
scheduler = APScheduler()
scheduler.init_app(flask)
scheduler.start()



@scheduler.task('interval', id='clean_email_verification_token_from_db', hours=12, misfire_grace_time=900)
def clean_email_verification_token_from_db():
    with flask.app_context():
        logging.info("Cleaning email verification tokens from database ...")
        from zero_totp_db_model.model import EmailVerificationToken
        
        tokens = db.session.query(EmailVerificationToken).all()
        count = 0
        for token in tokens:
            if float(token.expiration) < dt.datetime.now().timestamp():
                db.session.delete(token)
                db.session.commit()
                count += 1
        
        logging.info(f"Deleted {count} email verification tokens at {dt.datetime.now(dt.UTC).isoformat()}")

@scheduler.task('interval', id='clean_rate_limiting_from_db', hours=2, misfire_grace_time=900)
def clean_rate_limiting_from_db():
    with flask.app_context():
        logging.info("Cleaning rate limits from database ...")
        from database.rate_limiting_repo import RateLimitingRepo
        RateLimitingRepo().flush_outdated_limit()
        logging.info(f"Rate limits cleaned at {dt.datetime.now(dt.UTC).isoformat()}")


@scheduler.task('interval', id='clean_expired_refresh_token', hours=12, misfire_grace_time=900)
def clean_expired_refresh_token():
    with flask.app_context():
        logging.info("Cleaning expired refresh tokens from database ...")
        from zero_totp_db_model.model import RefreshToken
        tokens = db.session.query(RefreshToken).all()
        count=0
        minimum_retention_time = 24*60*60 # 24 hours
        for token in tokens:
            if float(token.expiration) + minimum_retention_time < dt.datetime.now(dt.UTC).timestamp():
                db.session.delete(token)
                db.session.commit()
                count += 1
        logging.info(f"Deleted {count} expired refresh tokens at {dt.datetime.now(dt.UTC).isoformat()}")


@scheduler.task('interval', id='clean_rate_limiting_from_db', hours=12, misfire_grace_time=900)
def clean_rate_limiting_from_db():
    with flask.app_context():
        logging.info("Cleaning expired session tokens from database ...")
        from zero_totp_db_model.model import SessionToken
        tokens = db.session.query(SessionToken).all()
        count=0
        minimum_retention_time = 24*60*60 # 24 hours
        for token in tokens:
            if float(token.expiration) + minimum_retention_time < dt.datetime.now(dt.UTC).timestamp():
                db.session.delete(token)
                db.session.commit()
                count += 1
        logging.info(f"Deleted {count} expired session tokens at {dt.datetime.now(dt.UTC).isoformat()}")



@flask.after_request
def after_request(response):
    from Utils import utils
    if response.status_code >= 500:
        logging.error(f"Completed request ip={utils.get_ip(request)} gw={request.remote_addr} url={request.url} method={request.method} status={response.status_code}")
    elif response.status_code >= 400:
        logging.warning(f"Completed request ip={utils.get_ip(request)} gw={request.remote_addr} url={request.url} method={request.method} status={response.status_code}")
    else :
        if request.path == "/api/v1/healthcheck":
            logging.debug(f"Completed healthcheck. If healthcheck failed, please check the logs above.")
        else:
            logging.info(f"Completed request ip={utils.get_ip(request)} gw={request.remote_addr} url={request.url} method={request.method} status={response.status_code}")
    return response
    


@flask.errorhandler(404)
def not_found(error):
    from Utils import utils
    logging.warning(f"error=404 ip={utils.get_ip(request)} gw={request.remote_addr} url={request.url}")
    return make_response(redirect(f"https://{conf.environment.domain}/404",  code=302))
            




if __name__ == "__main__":
   uvicorn.run("app:app", host="0.0.0.0", port=conf.api.port, reload=True)
   



1	import connexion	1✔
2	from flask_cors import CORS	1✔
3	from environment import conf	1✔
4	from database.db import db	1✔
5	from zero_totp_db_model.model_init import init_db	1✔
6	import uvicorn	1✔
7	from asgiref.wsgi import WsgiToAsgi	1✔
8	from starlette.middleware.cors import CORSMiddleware	1✔
9	from connexion.middleware import MiddlewarePosition	1✔
10	from environment import logging	1✔
11	import contextlib	1✔
12	from flask_apscheduler import APScheduler	1✔
13	from monitoring.sentry import sentry_configuration	1✔
14	from flask_migrate import Migrate	1✔
15	import datetime as dt	1✔
16	from flask import request, redirect, make_response	1✔
17
18
19	def create_app():	1✔
20	app_instance = connexion.FlaskApp(__name__, specification_dir="./openAPI/")	1✔
21	app_instance.add_api("swagger.yml")	1✔
22
23	app = app_instance.app	1✔
24
25	app.config["SQLALCHEMY_DATABASE_URI"] = conf.database.database_uri	1✔
26	app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False	1✔
27	app.config["PROPAGATE_EXCEPTIONS"] = True	1✔
28	app.secret_key = conf.api.flask_secret_key	1✔
29
30
31
32	db.init_app(app)	1✔
33	sentry_configuration() #optional	1✔
34	init_db(db)	1✔
35
36
37
38
39	return app_instance, app	1✔
40	app, flask = create_app()	1✔
41	migrate = Migrate(flask, db)	1✔
42	scheduler = APScheduler()	1✔
43	scheduler.init_app(flask)	1✔
44	scheduler.start()	1✔
45
46
47
48	@scheduler.task('interval', id='clean_email_verification_token_from_db', hours=12, misfire_grace_time=900)	1✔
49	def clean_email_verification_token_from_db():	1✔
50	with flask.app_context():	×
51	logging.info("Cleaning email verification tokens from database ...")	×
52	from zero_totp_db_model.model import EmailVerificationToken	×
53
54	tokens = db.session.query(EmailVerificationToken).all()	×
55	count = 0	×
56	for token in tokens:	×
57	if float(token.expiration) < dt.datetime.now().timestamp():	×
58	db.session.delete(token)	×
59	db.session.commit()	×
60	count += 1	×
61
62	logging.info(f"Deleted {count} email verification tokens at {dt.datetime.now(dt.UTC).isoformat()}")	×
63
64	@scheduler.task('interval', id='clean_rate_limiting_from_db', hours=2, misfire_grace_time=900)	1✔
65	def clean_rate_limiting_from_db():	1✔
66	with flask.app_context():	×
67	logging.info("Cleaning rate limits from database ...")	×
68	from database.rate_limiting_repo import RateLimitingRepo	×
69	RateLimitingRepo().flush_outdated_limit()	×
70	logging.info(f"Rate limits cleaned at {dt.datetime.now(dt.UTC).isoformat()}")	×
71
72
73	@scheduler.task('interval', id='clean_expired_refresh_token', hours=12, misfire_grace_time=900)	1✔
74	def clean_expired_refresh_token():	1✔
75	with flask.app_context():	×
76	logging.info("Cleaning expired refresh tokens from database ...")	×
77	from zero_totp_db_model.model import RefreshToken	×
78	tokens = db.session.query(RefreshToken).all()	×
79	count=0	×
80	minimum_retention_time = 246060 # 24 hours	×
81	for token in tokens:	×
82	if float(token.expiration) + minimum_retention_time < dt.datetime.now(dt.UTC).timestamp():	×
83	db.session.delete(token)	×
84	db.session.commit()	×
85	count += 1	×
86	logging.info(f"Deleted {count} expired refresh tokens at {dt.datetime.now(dt.UTC).isoformat()}")	×
87
88
89	@scheduler.task('interval', id='clean_rate_limiting_from_db', hours=12, misfire_grace_time=900)	1✔
90	def clean_rate_limiting_from_db():	1✔
91	with flask.app_context():	×
92	logging.info("Cleaning expired session tokens from database ...")	×
93	from zero_totp_db_model.model import SessionToken	×
94	tokens = db.session.query(SessionToken).all()	×
95	count=0	×
96	minimum_retention_time = 246060 # 24 hours	×
97	for token in tokens:	×
98	if float(token.expiration) + minimum_retention_time < dt.datetime.now(dt.UTC).timestamp():	×
99	db.session.delete(token)	×
100	db.session.commit()	×
101	count += 1	×
102	logging.info(f"Deleted {count} expired session tokens at {dt.datetime.now(dt.UTC).isoformat()}")	×
103
104
105
106	@flask.after_request	1✔
107	def after_request(response):	1✔
108	from Utils import utils	1✔
109	if response.status_code >= 500:	1✔
110	logging.error(f"Completed request ip={utils.get_ip(request)} gw={request.remote_addr} url={request.url} method={request.method} status={response.status_code}")	1✔
111	elif response.status_code >= 400:	1✔
112	logging.warning(f"Completed request ip={utils.get_ip(request)} gw={request.remote_addr} url={request.url} method={request.method} status={response.status_code}")	1✔
113	else :
114	if request.path == "/api/v1/healthcheck":	1✔
UNCOV 115	logging.debug(f"Completed healthcheck. If healthcheck failed, please check the logs above.")	×
116	else:
117	logging.info(f"Completed request ip={utils.get_ip(request)} gw={request.remote_addr} url={request.url} method={request.method} status={response.status_code}")	1✔
118	return response	1✔
119
120
121
122	@flask.errorhandler(404)	1✔
123	def not_found(error):	1✔
UNCOV 124	from Utils import utils	×
UNCOV 125	logging.warning(f"error=404 ip={utils.get_ip(request)} gw={request.remote_addr} url={request.url}")	×
UNCOV 126	return make_response(redirect(f"https://{conf.environment.domain}/404", code=302))	×
127
128
129
130
131
132	if __name__ == "__main__":	1✔
133	uvicorn.run("app:app", host="0.0.0.0", port=conf.api.port, reload=True)	×
134
135
136

SeaweedbrainCY / zero-totp / 12592571620

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous