How to use join？ about piccolo

[dantownsend]

Yeah, it's not super easy with Piccolo at the moment - it abstracts away joins, which helps in most situations, but can be harder in others.

I usually solve these problems using a sub select:

from piccolo.columns import Integer, Varchar, ForeignKey,Text
from piccolo.engine.postgres import PostgresEngine
from piccolo.table import Table, create_db_tables, drop_db_tables
from piccolo.query import SelectRaw


DB = PostgresEngine({'database': 'piccolo_left_join_test'})


class User(Table, db=DB):
    id = Integer(primary_key=True)
    username = Varchar()


class Posts(Table, db=DB):
    id = Integer(primary_key=True)
    poster = ForeignKey(references=User)
    title = Varchar()


class Comments(Table, db=DB):
    id = Integer(primary_key=True)
    commenter = ForeignKey(references=User)
    content = Text()


async def main():
    await drop_db_tables(User, Posts, Comments)
    await create_db_tables(User, Posts, Comments)

    ###########################################################################

    # Create some test data

    await User(id=1, username='Bob').save()
    await User(id=2, username='Alice').save()

    await Posts(id=1, poster=1, title='Bob post 1').save()
    await Posts(id=2, poster=1, title='Bob post 2').save()
    await Posts(id=3, poster=2, title='Alice post 1').save()

    await Comments(id=1, commenter=1, content='Bob comment 1').save()
    await Comments(id=2, commenter=1, content='Bob comment 2').save()
    await Comments(id=3, commenter=2, content='Alice comment 1').save()

    ###########################################################################

    # Example query
    response = await Posts.select(
        Posts.all_columns(),
        SelectRaw(
            "(SELECT COUNT(*) FROM comments WHERE commenter = posts.poster) AS comment_count"
        )
    )

    print(response)


if __name__ == '__main__':
    import asyncio
    asyncio.run(main())

[bymoye]

是的，目前 Piccolo 并不是那么容易 - 它抽象了连接，这在大多数情况下都有帮助，但在其他情况下可能会更困难。

我通常使用子选择来解决这些问题：

from piccolo.columns import Integer, Varchar, ForeignKey,Text
from piccolo.engine.postgres import PostgresEngine
from piccolo.table import Table, create_db_tables, drop_db_tables
from piccolo.query import SelectRaw


DB = PostgresEngine({'database': 'piccolo_left_join_test'})


class User(Table, db=DB):
    id = Integer(primary_key=True)
    username = Varchar()


class Posts(Table, db=DB):
    id = Integer(primary_key=True)
    poster = ForeignKey(references=User)
    title = Varchar()


class Comments(Table, db=DB):
    id = Integer(primary_key=True)
    commenter = ForeignKey(references=User)
    content = Text()


async def main():
    await drop_db_tables(User, Posts, Comments)
    await create_db_tables(User, Posts, Comments)

    ###########################################################################

    # Create some test data

    await User(id=1, username='Bob').save()
    await User(id=2, username='Alice').save()

    await Posts(id=1, poster=1, title='Bob post 1').save()
    await Posts(id=2, poster=1, title='Bob post 2').save()
    await Posts(id=3, poster=2, title='Alice post 1').save()

    await Comments(id=1, commenter=1, content='Bob comment 1').save()
    await Comments(id=2, commenter=1, content='Bob comment 2').save()
    await Comments(id=3, commenter=2, content='Alice comment 1').save()

    ###########################################################################

    # Example query
    response = await Posts.select(
        Posts.all_columns(),
        SelectRaw(
            "(SELECT COUNT(*) FROM comments WHERE commenter = posts.poster) AS comment_count"
        )
    )

    print(response)


if __name__ == '__main__':
    import asyncio
    asyncio.run(main())

Thanks for the reply, I think I know how I should encapsulate it