CalvinBackup/facefusion
1
0
Fork 0
mirror of https://github.com/facefusion/facefusion.git synced 2026-05-13 02:34:43 +02:00
Code Issues Packages Projects Releases Wiki Activity

Remove unused code to improve test coverage (#1050)

Browse Source 
* remove unused code

* move create_session_guard to middlewares
This commit is contained in:
Harisreedhar
2026-02-25 14:45:08 +05:30
committed by henryruhs
parent f69fa62fa2
commit 401118106a
5 changed files with 78 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
facefusion/apis/core.py
+2 -1
View File
@@ -8,8 +8,9 @@ from facefusion.apis.endpoints.capabilities import get_capabilities
from facefusion.apis.endpoints.metrics import get_metrics, websocket_metrics
from facefusion.apis.endpoints.ping import websocket_ping
from facefusion.apis.endpoints.process import websocket_process_image
from facefusion.apis.endpoints.session import create_session, create_session_guard, destroy_session, get_session, refresh_session
from facefusion.apis.endpoints.session import create_session, destroy_session, get_session, refresh_session
from facefusion.apis.endpoints.state import get_state, set_state
from facefusion.apis.middlewares.session import create_session_guard
def create_api() -> Starlette:
facefusion/apis/endpoints/session.py
+13 -85
View File
@@ -1,16 +1,12 @@
import os
import secrets
from typing import Optional
from starlette.datastructures import Headers
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.status import HTTP_200_OK, HTTP_201_CREATED, HTTP_401_UNAUTHORIZED, HTTP_426_UPGRADE_REQUIRED
from starlette.types import ASGIApp, Receive, Scope, Send
from starlette.status import HTTP_200_OK, HTTP_201_CREATED, HTTP_401_UNAUTHORIZED
from facefusion import session_context, session_manager, translator
from facefusion.apis.api_helper import get_sec_websocket_protocol
from facefusion.types import Token
from facefusion.apis.session_helper import extract_access_token
async def create_session(request : Request) -> JSONResponse:
@@ -36,25 +32,16 @@ async def create_session(request : Request) -> JSONResponse:
async def get_session(request : Request) -> JSONResponse:
access_token = extract_access_token(request.scope)
if access_token:
session_id = session_manager.find_session_id(access_token)
if session_id:
session = session_manager.get_session(session_id)
return JSONResponse(
{
'access_token': session.get('access_token'),
'refresh_token': session.get('refresh_token'),
'created_at': session.get('created_at').isoformat(),
'expires_at': session.get('expires_at').isoformat()
}, status_code = HTTP_200_OK)
session_id = session_manager.find_session_id(access_token)
session = session_manager.get_session(session_id)
return JSONResponse(
{
'message': translator.get('something_went_wrong', 'facefusion.apis')
}, status_code = HTTP_401_UNAUTHORIZED)
'access_token': session.get('access_token'),
'refresh_token': session.get('refresh_token'),
'created_at': session.get('created_at').isoformat(),
'expires_at': session.get('expires_at').isoformat()
}, status_code = HTTP_200_OK)
async def refresh_session(request : Request) -> JSONResponse:
@@ -79,69 +66,10 @@ async def refresh_session(request : Request) -> JSONResponse:
async def destroy_session(request : Request) -> JSONResponse:
access_token = extract_access_token(request.scope)
if access_token:
session_id = session_manager.find_session_id(access_token)
if session_id:
session_manager.clear_session(session_id)
return JSONResponse(
{
'message': translator.get('ok', 'facefusion.apis')
}, status_code = HTTP_200_OK)
session_id = session_manager.find_session_id(access_token)
session_manager.clear_session(session_id)
return JSONResponse(
{
'message': translator.get('something_went_wrong', 'facefusion.apis')
}, status_code = HTTP_401_UNAUTHORIZED)
def create_session_guard(app : ASGIApp) -> ASGIApp:
async def middleware(scope : Scope, receive : Receive, send : Send) -> None:
access_token = extract_access_token(scope)
if access_token:
session_id = session_manager.find_session_id(access_token)
if session_id:
if session_manager.validate_session(session_id):
return await app(scope, receive, send)
response = JSONResponse(
{
'message': translator.get('invalid_access_token', 'facefusion.apis')
}, status_code = HTTP_426_UPGRADE_REQUIRED)
return await response(scope, receive, send)
response = JSONResponse(
{
'message': translator.get('invalid_access_token', 'facefusion.apis')
}, status_code = HTTP_401_UNAUTHORIZED)
return await response(scope, receive, send)
return middleware
def extract_access_token(scope : Scope) -> Optional[Token]:
if scope.get('type') == 'http':
auth_header = Headers(scope = scope).get('Authorization')
if auth_header:
auth_prefix, _, access_token = auth_header.partition(' ')
if auth_prefix.lower() == 'bearer' and access_token:
return access_token
if scope.get('type') == 'websocket':
subprotocol = get_sec_websocket_protocol(scope)
if subprotocol:
protocol_prefix, _, access_token = subprotocol.partition('.')
if protocol_prefix == 'access_token' and access_token:
return access_token
return None
'message': translator.get('ok', 'facefusion.apis')
}, status_code = HTTP_200_OK)
facefusion/apis/middlewares/__init__.py
View File
facefusion/apis/middlewares/session.py
+34
View File
@@ -0,0 +1,34 @@
from starlette.responses import JSONResponse
from starlette.status import HTTP_401_UNAUTHORIZED, HTTP_426_UPGRADE_REQUIRED
from starlette.types import ASGIApp, Receive, Scope, Send
from facefusion import session_manager, translator
from facefusion.apis.session_helper import extract_access_token
def create_session_guard(app : ASGIApp) -> ASGIApp:
async def middleware(scope : Scope, receive : Receive, send : Send) -> None:
access_token = extract_access_token(scope)
if access_token:
session_id = session_manager.find_session_id(access_token)
if session_id:
if session_manager.validate_session(session_id):
return await app(scope, receive, send)
response = JSONResponse(
{
'message': translator.get('invalid_access_token', 'facefusion.apis')
}, status_code = HTTP_426_UPGRADE_REQUIRED)
return await response(scope, receive, send)
response = JSONResponse(
{
'message': translator.get('invalid_access_token', 'facefusion.apis')
}, status_code = HTTP_401_UNAUTHORIZED)
return await response(scope, receive, send)
return middleware
facefusion/apis/session_helper.py
+29
View File
@@ -0,0 +1,29 @@
from typing import Optional
from starlette.datastructures import Headers
from starlette.types import Scope
from facefusion.apis.api_helper import get_sec_websocket_protocol
from facefusion.types import Token
def extract_access_token(scope : Scope) -> Optional[Token]:
if scope.get('type') == 'http':
auth_header = Headers(scope = scope).get('Authorization')
if auth_header:
auth_prefix, _, access_token = auth_header.partition(' ')
if auth_prefix.lower() == 'bearer' and access_token:
return access_token
if scope.get('type') == 'websocket':
subprotocol = get_sec_websocket_protocol(scope)
if subprotocol:
protocol_prefix, _, access_token = subprotocol.partition('.')
if protocol_prefix == 'access_token' and access_token:
return access_token
return None