Skip to content

Client Storage DAO

Module: s2auth.client.dao

Base = declarative_base() module-attribute

ConnectionDetail

Bases: Base

Source code in src/s2auth/client/dao.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class ConnectionDetail(Base):
    __tablename__ = "connection_details"

    s2_node_id: Mapped[str] = mapped_column(String, nullable=False, index=True, primary_key=True)
    client_s2_node_id: Mapped[Optional[str]] = mapped_column(String, nullable=True)
    pairing_server_url: Mapped[Optional[str]] = mapped_column(String, nullable=True)
    verify_tls: Mapped[Optional[bool]] = mapped_column(Boolean, nullable=True)
    ssl_certfile: Mapped[Optional[str]] = mapped_column(String, nullable=True)
    initiate_session_url: Mapped[Optional[str]] = mapped_column(String, nullable=True)
    access_token: Mapped[Optional[str]] = mapped_column(String, nullable=True)
    supported_s2_message_versions: Mapped[Optional[list[str]]] = mapped_column(JSON, nullable=True)
    supported_communication_protocols: Mapped[Optional[list[str]]] = mapped_column(JSON, nullable=True)
    supported_hmac_hashing_algorithms: Mapped[Optional[list[str]]] = mapped_column(JSON, nullable=True)
    selected_s2_message_version: Mapped[Optional[str]] = mapped_column(String, nullable=True)
    selected_communication_protocol: Mapped[Optional[str]] = mapped_column(String, nullable=True)
    selected_hmac_hashing_algorithm: Mapped[Optional[str]] = mapped_column(String, nullable=True)
    server_node_description: Mapped[Optional[dict[str, Any]]] = mapped_column(JSON, nullable=True)
    server_endpoint_description: Mapped[Optional[dict[str, Any]]] = mapped_column(JSON, nullable=True)

Dao

SQLAlchemy-backed data access object for storing/loading connection details. Default database is SQLite file 'connection_details.db'.

Source code in src/s2auth/client/dao.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
class Dao:
    """
    SQLAlchemy-backed data access object for storing/loading connection details.
    Default database is SQLite file 'connection_details.db'.
    """

    def __init__(self, db_url: str = "sqlite:///connection_details.db") -> None:
        # Create engine & session factory
        self._engine = create_engine(db_url, future=True)
        Base.metadata.create_all(self._engine)

        self._SessionLocal = sessionmaker(
            bind=self._engine,
            autoflush=False,
            autocommit=False,
            future=True,
        )

    def store_connection_details(self, s2_node_id: str, details: dict[str, Any]) -> None:
        """
        Insert or overwrite a connection detail identified by s2_node_id.
        """
        with self._SessionLocal() as session:
            with session.begin():
                obj = session.query(ConnectionDetail).filter(ConnectionDetail.s2_node_id == s2_node_id).one_or_none()

                if obj is None:
                    obj = ConnectionDetail(s2_node_id=s2_node_id)
                    session.add(obj)

                for detail_key, model_attr in details.items():
                    if hasattr(ConnectionDetail, detail_key):
                        setattr(obj, detail_key, model_attr)
                    else:
                        raise ValueError(f"Invalid detail key: {detail_key}")

    def load_connection_details(self, s2_node_id: str) -> Optional[dict[str, Any]]:
        """Load the full connection details object for the given node ID."""
        stmt: Select[Any] = (
            select(ConnectionDetail)
            .where(ConnectionDetail.s2_node_id == s2_node_id)
            .limit(1)
        )
        with self._SessionLocal() as session:
            obj = session.execute(stmt).scalars().first()
            if obj is None:
                return None
            return {c.key: getattr(obj, c.key) for c in inspect(obj).mapper.column_attrs}

    def remove_connection_details(self, s2_node_id: str) -> bool:
        """Remove connection details for the given node ID.

        Returns True when an entry was deleted, otherwise False.
        """
        stmt: Select[Any] = (
            select(ConnectionDetail)
            .where(ConnectionDetail.s2_node_id == s2_node_id)
            .limit(1)
        )
        with self._SessionLocal() as session:
            with session.begin():
                obj = session.execute(stmt).scalars().first()
                if obj is None:
                    return False
                session.delete(obj)
            return True

store_connection_details(s2_node_id, details)

Insert or overwrite a connection detail identified by s2_node_id.

Source code in src/s2auth/client/dao.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def store_connection_details(self, s2_node_id: str, details: dict[str, Any]) -> None:
    """
    Insert or overwrite a connection detail identified by s2_node_id.
    """
    with self._SessionLocal() as session:
        with session.begin():
            obj = session.query(ConnectionDetail).filter(ConnectionDetail.s2_node_id == s2_node_id).one_or_none()

            if obj is None:
                obj = ConnectionDetail(s2_node_id=s2_node_id)
                session.add(obj)

            for detail_key, model_attr in details.items():
                if hasattr(ConnectionDetail, detail_key):
                    setattr(obj, detail_key, model_attr)
                else:
                    raise ValueError(f"Invalid detail key: {detail_key}")

load_connection_details(s2_node_id)

Load the full connection details object for the given node ID.

Source code in src/s2auth/client/dao.py 
68
69
70
71
72
73
74
75
76
77
78
79
def load_connection_details(self, s2_node_id: str) -> Optional[dict[str, Any]]:
    """Load the full connection details object for the given node ID."""
    stmt: Select[Any] = (
        select(ConnectionDetail)
        .where(ConnectionDetail.s2_node_id == s2_node_id)
        .limit(1)
    )
    with self._SessionLocal() as session:
        obj = session.execute(stmt).scalars().first()
        if obj is None:
            return None
        return {c.key: getattr(obj, c.key) for c in inspect(obj).mapper.column_attrs}

remove_connection_details(s2_node_id)

Remove connection details for the given node ID.

Returns True when an entry was deleted, otherwise False.

Source code in src/s2auth/client/dao.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def remove_connection_details(self, s2_node_id: str) -> bool:
    """Remove connection details for the given node ID.

    Returns True when an entry was deleted, otherwise False.
    """
    stmt: Select[Any] = (
        select(ConnectionDetail)
        .where(ConnectionDetail.s2_node_id == s2_node_id)
        .limit(1)
    )
    with self._SessionLocal() as session:
        with session.begin():
            obj = session.execute(stmt).scalars().first()
            if obj is None:
                return False
            session.delete(obj)
        return True