SQLalchemy：将经典映射转换为命令式

Question

我正在与一个团队合作开发一个项目，该项目使用 SQLAlchemy 与我们的后端数据库进行通信。我以经典的方式编写了代码，但由于某种原因，这种方式不受欢迎，而且我很难理解命令式的映射方式。这些是我必须通过命令式方式重写的方法：

    def get_wishlist_games(self, user: User):
        # totally just avoid the stupid user object.
        print('test')
        if user and user.user_id:
            user_id = user.user_id
            # Query the wishlist_id associated with the user
            wishlist_query = select([wishlists_table.c.wishlist_id]).where(wishlists_table.c.user_id == user_id)
            wishlist = self._session_cm.session.execute(wishlist_query).fetchone()

            if wishlist:
                print('inside wishlist')
                wishlist_id = wishlist['wishlist_id']
                # Query the game_wishlists_table to get game_ids associated with the wishlist_id
                game_ids_query = select([game_wishlists_table.c.game_id]).where(
                    game_wishlists_table.c.wishlist_id == wishlist_id
                )
                game_ids = self._session_cm.session.execute(game_ids_query).fetchall()

                # Fetch the games using the game_ids
                game_list = []
                for game_id in game_ids:
                    game = self.get_game_by_id(game_id[0])
                    if game:
                        game_list.append(game)
                return game_list

        return []

    def check_game_in_wishlist(self, user: User, game: Game):
        # Get the user's wishlist ID (assuming it exists)
        user_id = user.user_id
        wishlist_id_query = select([wishlists_table.c.wishlist_id]).where(wishlists_table.c.user_id == user_id)
        wishlist_id = self._session_cm.session.execute(wishlist_id_query).scalar()

        if wishlist_id is not None:
            # Check if the game is in the user's wishlist
            game_wishlist_query = select([game_wishlists_table.c.game_id]).where(
                (game_wishlists_table.c.game_id == game.game_id) &
                (game_wishlists_table.c.wishlist_id == wishlist_id)
            )
            game_id = self._session_cm.session.execute(game_wishlist_query).scalar()

            return game_id is not None
        else:
            return False

    def add_game_to_wishlist(self, user: User, game: Game):
        with self._session_cm as scm:
            ##check if the user has an existing wishlist
            user_id = user.user_id
            wishlist_query = wishlists_table.select().where(wishlists_table.c.user_id == user_id)
            wishlist = scm.session.execute(wishlist_query).fetchone()

            if wishlist is None:
                # if no wishlist inside, makes a fresh one
                insert_wishlist_statement = wishlists_table.insert().values(user_id=user_id)
                result = scm.session.execute(insert_wishlist_statement)
                wishlist_id = result.inserted_primary_key[0]
            else:
                wishlist_id = wishlist['wishlist_id']

            #associate the game with the wishlist in the assoc table
            insert_game_wishlist_statement = game_wishlists_table.insert().values(game_id=game.game_id,
                                                                                  wishlist_id=wishlist_id)
            scm.session.execute(insert_game_wishlist_statement)

            scm.commit()

    def remove_game_from_wishlist(self, user: User, game_title: str):
        with self._session_cm as scm:
            print('test')
            # Find the user's wishlist by user_id
            user_id = user.user_id
            wishlist_query = select([wishlists_table.c.wishlist_id]).where(wishlists_table.c.user_id == user_id)
            wishlist = scm.session.execute(wishlist_query).fetchone()

            print('outside')
            if wishlist is not None:
                print('inside')
                wishlist_id = wishlist['wishlist_id']

                # Find the game_wishlist_assoc entry with the matching game title and wishlist_id
                game_wishlist_query = select([game_wishlists_table]).where(
                    (game_wishlists_table.c.game_id == game_title) &
                    (game_wishlists_table.c.wishlist_id == wishlist_id)
                )
                game_wishlist = scm.session.execute(game_wishlist_query).fetchone()

                print(game_wishlist)

                if game_wishlist is not None:
                    # Remove the game_wishlist_assoc entry to remove the game from the wishlist
                    delete_statement = game_wishlists_table.delete().where(
                        (game_wishlists_table.c.game_id == game_title) &
                        (game_wishlists_table.c.wishlist_id == wishlist_id)
                    )
                    scm.session.execute(delete_statement)
                    scm.commit()

此外，我使用的ORM如下：

from sqlalchemy import (
    Table, MetaData, Column, Integer, String, Text, Float, ForeignKey, DateTime
)
from sqlalchemy.orm import relationship, registry

from games.domainmodel.model import Game, Publisher, Genre, User, Review, Wishlist

# global variable giving access to the MetaData (schema) information of the database
metadata = MetaData()

mapper_registry = registry()


publishers_table = Table(
    'publishers', metadata,
    Column('name', String(255), primary_key=True)  # nullable=False, unique=True)
)

games_table = Table(
    'games', metadata,
    Column('game_id', Integer, primary_key=True, nullable=False),
    Column('title', Text, nullable=False),
    Column('price', Float, nullable=False),
    Column('release_date', String(50), nullable=False),
    Column('description', String(255), nullable=True),
    Column('image_url', String(255), nullable=True),
    Column('website_url', String(255), nullable=True),
    Column('publisher_name', ForeignKey('publishers.name'))
)

genres_table = Table(
    'genres', metadata,
    Column('genre_name', String(64), primary_key=True, nullable=False)
)

game_genres_table = Table(
    'game_genres', metadata,
    Column('id', Integer, primary_key=True, autoincrement=True),
    Column('game_id', ForeignKey('games.game_id')),
    Column('genre_name', ForeignKey('genres.genre_name'))
)

users_table = Table(
    'users', metadata,
    Column('user_id', Integer, primary_key=True, autoincrement=True),
    Column('username', String(20), unique=True, nullable=False),
    Column('password', String(20), nullable=False)
)

reviews_table = Table(
    'reviews', metadata,
    Column('review_id', Integer, primary_key=True, autoincrement=True),
    Column('user_id', ForeignKey('users.user_id')),
    Column('game_id', ForeignKey('games.game_id')),
    Column('comment', String(255), nullable=False),
    Column('rating', Integer, nullable=False),
    Column('timestamp', DateTime, nullable=False)
)

wishlists_table = Table(
    'wishlists', metadata,
    Column('wishlist_id', Integer, primary_key=True, autoincrement=True),
    Column('user_id', ForeignKey('users.user_id'))
)

game_wishlists_table = Table(
    'game_wishlists', metadata,
    Column('id', Integer, primary_key=True, autoincrement=True),
    Column('game_id', ForeignKey('games.game_id')),
    Column('wishlist_id', ForeignKey('wishlists.wishlist_id'))
)


def map_model_to_tables():
    mapper_registry.map_imperatively(Publisher, publishers_table, properties={
        '_Publisher__publisher_name': publishers_table.c.name,
    })

    mapper_registry.map_imperatively(Game, games_table, properties={
        '_Game__game_id': games_table.c.game_id,
        '_Game__game_title': games_table.c.title,
        '_Game__price': games_table.c.price,
        '_Game__release_date': games_table.c.release_date,
        '_Game__description': games_table.c.description,
        '_Game__image_url': games_table.c.image_url,
        '_Game__website_url': games_table.c.website_url,
        '_Game__publisher': relationship(Publisher),
        '_Game__genres': relationship(Genre, secondary=game_genres_table),
        '_Game__reviews': relationship(Review, back_populates='_Review__game'),
    })

    mapper_registry.map_imperatively(Genre, genres_table, properties={
        '_Genre__genre_name': genres_table.c.genre_name,
    })

    mapper_registry.map_imperatively(User, users_table, properties={
        '_User__username': users_table.c.username,
        '_User__password': users_table.c.password,
        '_User__reviews': relationship(Review, back_populates='_Review__user'),
    })

    mapper_registry.map_imperatively(Review, reviews_table, properties={
        '_Review__rating': reviews_table.c.rating,
        '_Review__comment': reviews_table.c.comment,
        '_Review__game': relationship(Game, back_populates='_Game__reviews'),
        '_Review__user': relationship(User, back_populates='_User__reviews'),
    })

    mapper_registry.map_imperatively(Wishlist, wishlists_table, properties={
        '_Wishlist__user': relationship(User),
        '_Wishlist__list_of_games': relationship(Game, secondary=game_wishlists_table),
    })

我应该操作的相关数据库如下：

与此数据库有关系：

Answer 1

这是转换为命令式样式的代码：

 def get_wishlist_games(self, user: User):
        game_list = []

        if user and user.user_id:
            with self._session_cm as scm:
                session = scm.session

                # Query the wishlist associated with the user
                user_id = user.user_id
                wishlist_query = session.query(Wishlist).filter(Wishlist.user_id == user_id).first()

                if wishlist_query:
                    # Fetch the games associated with the wishlist
                    game_ids = [game_wishlist.game_id for game_wishlist in wishlist_query.list_of_games]
                    game_list = [self.get_game_by_id(game_id) for game_id in game_ids if self.get_game_by_id(game_id)]

        return game_list


    def check_game_in_wishlist(self, user: User, game: Game):
        if user and user.user_id:
            with self._session_cm as scm:
                session = scm.session

                # Query the wishlist associated with the user
                user_id = user.user_id
                wishlist_query = session.query(Wishlist).filter(Wishlist.user_id == user_id).first()

                if wishlist_query:
                    # Check if the game is in the user's wishlist by searching within the list_of_games
                    return game in wishlist_query.list_of_games

        return False

    def add_game_to_wishlist(self, user: User, game: Game):
        if user and user.user_id:
            with self._session_cm as scm:
                session = scm.session

                # Query the wishlist associated with the user
                user_id = user.user_id
                wishlist_query = session.query(Wishlist).filter(Wishlist.user_id == user_id).first()

                if not wishlist_query:
                    # If no wishlist exists for the user, create a new one
                    wishlist_query = Wishlist(user_id=user_id)
                    session.add(wishlist_query)

                # Add the game to the user's wishlist
                wishlist_query.list_of_games.append(game)

                scm.commit()

    def remove_game_from_wishlist(self, user: User, game_title: str):
        if user and user.user_id:
            with self._session_cm as scm:
                session = scm.session

                # Query the wishlist associated with the user
                user_id = user.user_id
                wishlist_query = session.query(Wishlist).filter(Wishlist.user_id == user_id).first()

                if wishlist_query:
                    # Find the game object with the matching title within the user's wishlist
                    game_to_remove = next((game for game in wishlist_query.list_of_games if game.title == game_title), None)

                    if game_to_remove:
                        # Remove the game from the user's wishlist
                        wishlist_query.list_of_games.remove(game_to_remove)

                scm.commit()