♻️ Use declarative table configuration to define backend db tables

This commit is contained in:
2023-05-08 13:17:09 +02:00
parent c84b78d3e2
commit 93047c8706
3 changed files with 35 additions and 32 deletions

View File

@@ -5,13 +5,11 @@ from typing import Iterable, Self, Sequence
from sqlalchemy import ( from sqlalchemy import (
BigInteger, BigInteger,
Boolean, Boolean,
Column,
Enum, Enum,
ForeignKey, ForeignKey,
Integer, Integer,
select, select,
String, String,
Table,
) )
from sqlalchemy.orm import Mapped, mapped_column, relationship, selectinload from sqlalchemy.orm import Mapped, mapped_column, relationship, selectinload
from sqlalchemy.sql.expression import tuple_ from sqlalchemy.sql.expression import tuple_
@@ -25,12 +23,14 @@ from ..idfm_interface.idfm_types import (
) )
from .stop import _Stop from .stop import _Stop
line_stop_association_table = Table(
"line_stop_association_table", class LineStopAssociations(Base):
Base.metadata,
Column("line_id", ForeignKey("lines.id")), id = mapped_column(BigInteger, primary_key=True)
Column("stop_id", ForeignKey("_stops.id")), line_id = mapped_column(String, ForeignKey("lines.id"))
) stop_id = mapped_column(BigInteger, ForeignKey("_stops.id"))
__tablename__ = "line_stop_associations"
class LinePicto(Base): class LinePicto(Base):
@@ -83,7 +83,7 @@ class Line(Base):
stops: Mapped[list[_Stop]] = relationship( stops: Mapped[list[_Stop]] = relationship(
"_Stop", "_Stop",
secondary=line_stop_association_table, secondary="line_stop_associations",
back_populates="lines", back_populates="lines",
lazy="selectin", lazy="selectin",
) )

View File

@@ -1,11 +1,10 @@
from __future__ import annotations from __future__ import annotations
from logging import getLogger from logging import getLogger
from typing import Annotated, Iterable, Sequence, TYPE_CHECKING from typing import Iterable, Sequence, TYPE_CHECKING
from sqlalchemy import ( from sqlalchemy import (
BigInteger, BigInteger,
Column,
Computed, Computed,
desc, desc,
Enum, Enum,
@@ -16,7 +15,6 @@ from sqlalchemy import (
JSON, JSON,
select, select,
String, String,
Table,
) )
from sqlalchemy.orm import ( from sqlalchemy.orm import (
mapped_column, mapped_column,
@@ -37,12 +35,13 @@ if TYPE_CHECKING:
logger = getLogger(__name__) logger = getLogger(__name__)
stop_area_stop_association_table = Table( class StopAreaStopAssociations(Base):
"stop_area_stop_association_table",
Base.metadata, id = mapped_column(BigInteger, primary_key=True)
Column("stop_id", ForeignKey("_stops.id")), stop_id = mapped_column(BigInteger, ForeignKey("_stops.id"))
Column("stop_area_id", ForeignKey("stop_areas.id")), stop_area_id = mapped_column(BigInteger, ForeignKey("stop_areas.id"))
)
__tablename__ = "stop_area_stop_associations"
class _Stop(Base): class _Stop(Base):
@@ -64,12 +63,14 @@ class _Stop(Base):
lines: Mapped[list[Line]] = relationship( lines: Mapped[list[Line]] = relationship(
"Line", "Line",
secondary="line_stop_association_table", secondary="line_stop_associations",
back_populates="stops", back_populates="stops",
lazy="selectin", lazy="selectin",
) )
areas: Mapped[list["StopArea"]] = relationship( areas: Mapped[list["StopArea"]] = relationship(
"StopArea", secondary=stop_area_stop_association_table, back_populates="stops" "StopArea",
secondary="stop_area_stop_associations",
back_populates="stops",
) )
connection_area_id: Mapped[int] = mapped_column( connection_area_id: Mapped[int] = mapped_column(
ForeignKey("connection_areas.id"), nullable=True ForeignKey("connection_areas.id"), nullable=True
@@ -143,7 +144,7 @@ class StopArea(_Stop):
stops: Mapped[list["Stop"]] = relationship( stops: Mapped[list["Stop"]] = relationship(
"Stop", "Stop",
secondary=stop_area_stop_association_table, secondary="stop_area_stop_associations",
back_populates="areas", back_populates="areas",
lazy="selectin", lazy="selectin",
) )

View File

@@ -1,25 +1,27 @@
from sqlalchemy import Column, ForeignKey, String, Table from sqlalchemy import BigInteger, ForeignKey, String
from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.orm import Mapped, mapped_column, relationship
from ..db import Base, db from ..db import Base, db
from .stop import _Stop from .stop import _Stop
user_last_stop_search_stops_associations_table = Table(
"user_last_stop_search_stops_associations_table", class UserLastStopSearchStopAssociations(Base):
Base.metadata, id = mapped_column(BigInteger, primary_key=True)
Column("user_mxid", ForeignKey("user_last_stop_search_results.user_mxid")), user_mxid = mapped_column(
Column("stop_id", ForeignKey("_stops.id")), String, ForeignKey("user_last_stop_search_results.user_mxid")
) )
stop_id = mapped_column(BigInteger, ForeignKey("_stops.id"))
__tablename__ = "user_last_stop_search_stop_associations"
class UserLastStopSearchResults(Base): class UserLastStopSearchResults(Base):
db = db db = db
__tablename__ = "user_last_stop_search_results"
user_mxid = mapped_column(String, primary_key=True) user_mxid = mapped_column(String, primary_key=True)
request_content = mapped_column(String, nullable=False) request_content = mapped_column(String, nullable=False)
stops: Mapped[_Stop] = relationship( stops: Mapped[_Stop] = relationship(
_Stop, secondary=user_last_stop_search_stops_associations_table _Stop, secondary="user_last_stop_search_stop_associations"
) )
__tablename__ = "user_last_stop_search_results"