From bb4259bf24cd020ff3d247d105c2115375af5234 Mon Sep 17 00:00:00 2001 From: Giuseppe Papallo Date: Tue, 16 Apr 2024 08:34:24 -0300 Subject: [PATCH] replace pipes in type annotations with Optional --- src/flask_session/postgresql/postgresql.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/flask_session/postgresql/postgresql.py b/src/flask_session/postgresql/postgresql.py index e85bd537..b5f1e8d6 100644 --- a/src/flask_session/postgresql/postgresql.py +++ b/src/flask_session/postgresql/postgresql.py @@ -2,7 +2,7 @@ from contextlib import contextmanager from datetime import timedelta as TimeDelta -from typing import Generator +from typing import Generator, Optional from flask import Flask from itsdangerous import want_bytes @@ -40,7 +40,7 @@ class PostgreSqlSessionInterface(ServerSideSessionInterface): def __init__( self, app: Flask, - pool: ThreadedConnectionPool | None = Defaults.SESSION_POSTGRESQL, + pool: Optional[ThreadedConnectionPool] = Defaults.SESSION_POSTGRESQL, key_prefix: str = Defaults.SESSION_KEY_PREFIX, use_signer: bool = Defaults.SESSION_USE_SIGNER, permanent: bool = Defaults.SESSION_PERMANENT, @@ -48,7 +48,7 @@ def __init__( serialization_format: str = Defaults.SESSION_SERIALIZATION_FORMAT, table: str = Defaults.SESSION_POSTGRESQL_TABLE, schema: str = Defaults.SESSION_POSTGRESQL_SCHEMA, - cleanup_n_requests: int | None = Defaults.SESSION_CLEANUP_N_REQUESTS, + cleanup_n_requests: Optional[int] = Defaults.SESSION_CLEANUP_N_REQUESTS, ) -> None: if not isinstance(pool, ThreadedConnectionPool): raise TypeError("No valid ThreadedConnectionPool instance provided.") @@ -74,7 +74,7 @@ def __init__( @contextmanager def _get_cursor( - self, conn: PsycoPg2Connection | None = None + self, conn: Optional[PsycoPg2Connection] = None ) -> Generator[PsycoPg2Cursor, None, None]: _conn: PsycoPg2Connection = conn or self.pool.getconn() @@ -107,7 +107,7 @@ def _delete_session(self, store_id: str) -> None: ) @retry_query(max_attempts=3) - def _retrieve_session_data(self, store_id: str) -> dict | None: + def _retrieve_session_data(self, store_id: str) -> Optional[dict]: with self._get_cursor() as cur: cur.execute( self._queries.retrieve_session_data,