Code: Select all
async def get_by_title(session: AsyncSession, title: str) -> Optional[Manga]:
query = select(Manga).where(Manga.title == title).limit(1)
result = await session.scalar(query)
return result
Code: Select all
from .base import Base
from datetime import datetime
from sqlalchemy import func
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy.dialects.sqlite import INTEGER, TEXT, DATETIME
class Manga(Base):
__tablename__ = 'manga'
manga_id: Mapped[int] = mapped_column(INTEGER,primary_key=True,autoincrement=True)
title: Mapped[str] = mapped_column(TEXT,nullable=False)
language: Mapped[str] = mapped_column(TEXT,nullable=False)
path_key: Mapped[str] = mapped_column(TEXT,nullable=False)
created_at: Mapped[datetime] = mapped_column(DATETIME,nullable=False,default=func.now())
Code: Select all
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
pass
Code: Select all
async def print_rows(session: AsyncSession, title: str):
query = select(Manga).where(Manga.title == title)
result = await session.execute(query)
for row in result:
print(row) # (manga_id,title,language,path_key,created_at) all as expected