87 lines
2.4 KiB
Python
87 lines
2.4 KiB
Python
|
import asyncio
|
||
|
|
|
||
|
|
from sqlalchemy import select, Result
|
||
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
|
from sqlalchemy.orm import joinedload
|
||
|
|
from core.models import db_helper, User, Profile, Post
|
||
|
|
|
||
|
|
import logging
|
||
|
|
|
||
|
|
log = logging.getLogger()
|
||
|
|
|
||
|
|
|
||
|
|
async def create_user(session: AsyncSession, username) -> User:
|
||
|
|
user = User(username=username)
|
||
|
|
session.add(user)
|
||
|
|
await session.commit()
|
||
|
|
log.info("user %r", user)
|
||
|
|
return user
|
||
|
|
|
||
|
|
|
||
|
|
async def get_user_by_username(session: AsyncSession, username: str) -> User | None:
|
||
|
|
stmt = select(User).where(User.username == username)
|
||
|
|
# result: Result = await session.execute(stmt)
|
||
|
|
# user: User | None = result.scalar_one_or_none()
|
||
|
|
user: User | None = await session.scalar(stmt)
|
||
|
|
log.info("found user %r %r", username, user)
|
||
|
|
return user
|
||
|
|
|
||
|
|
|
||
|
|
async def create_user_profile(
|
||
|
|
session: AsyncSession,
|
||
|
|
user_id: int,
|
||
|
|
first_name: str | None = None,
|
||
|
|
last_name: str | None = None,
|
||
|
|
) -> Profile:
|
||
|
|
profile = Profile(
|
||
|
|
user_id=user_id,
|
||
|
|
first_name=first_name,
|
||
|
|
last_name=last_name,
|
||
|
|
)
|
||
|
|
session.add(profile)
|
||
|
|
await session.commit()
|
||
|
|
return profile
|
||
|
|
|
||
|
|
|
||
|
|
async def show_users_with_profiles(session: AsyncSession) -> list[User]:
|
||
|
|
stmt = select(User).options(joinedload(User.profile)).order_by(User.id)
|
||
|
|
users = await session.scalars(stmt)
|
||
|
|
for user in users:
|
||
|
|
log.info("user: %r profile first name: %r", user, user.profile.first_name)
|
||
|
|
|
||
|
|
|
||
|
|
async def create_posts(
|
||
|
|
session: AsyncSession,
|
||
|
|
user_id: int,
|
||
|
|
*posts_titles: str,
|
||
|
|
) -> list[Post]:
|
||
|
|
posts = [Post(title=title, user_id=user_id) for title in posts_titles]
|
||
|
|
session.add_all(posts)
|
||
|
|
await session.commit()
|
||
|
|
log.info("posts: %r", posts)
|
||
|
|
return posts
|
||
|
|
|
||
|
|
|
||
|
|
async def main():
|
||
|
|
async with db_helper.session_factory() as session:
|
||
|
|
# await create_user(session=session, username="sergey")
|
||
|
|
# await create_user(session=session, username="vasya")
|
||
|
|
user_vasya = await get_user_by_username(session=session, username="vasya")
|
||
|
|
# # await get_user_by_username(session=session, username="sefesf")
|
||
|
|
# await create_user_profile(
|
||
|
|
# session=session,
|
||
|
|
# user_id=user_vasya.id,
|
||
|
|
# first_name="вася",
|
||
|
|
# )
|
||
|
|
# await show_users_with_profiles(session=session)
|
||
|
|
await create_posts(
|
||
|
|
session,
|
||
|
|
user_vasya.id,
|
||
|
|
"sqla 2.0",
|
||
|
|
"sqla joins",
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
asyncio.run(main())
|