Wie überprüfe ich den Antworttyp in der Middleware? ⇐ Python

[Anonymous]

Ich versuche eine Middleware zu erstellen, deren Logik vom Antworttyp abhängt. Insbesondere muss ich die DB -Sitzung schließen, nachdem die Antwort für "sofortige" Antworttypen (wie Antwort , htmlresponse , jsonResponse gesendet wurde, und muss nicht schließen, um es zu streamingResponse zu schließen, da ich die Verbindung schließen muss, nachdem der Stream fertig ist. Ich drucke Typ (Antwort) , ich gene 'starlette.middleware.base._streamingResponse' , unabhängig von dem Antworttyp, den ich im Router eingestellt habe. Klasse?
Ist es überhaupt möglich, den Antworttyp zu überprüfen? Beispiel: < /p>

Code: Select all

_engine = create_async_engine(
f"postgresql+asyncpg://{user}:{password}@{host}:{port}/{database}"
)
AsyncSessionFactory = async_sessionmaker(bind=_engine, expire_on_commit=False)

class DBSessionMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
request.state.db_session_factory = AsyncSessionFactory
request.state.active_session = None

response = await call_next(request)
debug_logger.debug(f'response type: {type(response)}')  # always prints `StreamingResponse`

if request.state.active_session and not isinstance(response, StreamingResponse):
request.state.active_session.close()

return response

# get or create the sesson from the factory
async def get_db_session(request: Request) -> AsyncSession:
if hasattr(request.state, "active_session") and request.state.active_session:
session = request.state.active_session
elif hasattr(request.state, "db_session_factory"):
request.state.active_session = request.state.db_session_factory()
session = request.state.active_session
debug_logger.debug(f'session {id(session)} opened')
else:
session = AsyncSessionFactory()
return cast(AsyncSession, session)