update test cases

Author: wu-clan

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "sqlalchemy-crud-plus"
 description = "Asynchronous CRUD operation based on SQLAlchemy2 model"
-version = '1.9.0'
+version = '1.10.0'
 authors = [
     { name = "Wu Clan", email = "jianhengwu0407@gmail.com" },
 ]

requirements_docs.txt

@@ -1 +1,2 @@
-mkdocs-material==9.5.31
+mkdocs-material
+mkdocstrings[python]

sqlalchemy_crud_plus/types.py

@@ -1,3 +1,5 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
 from __future__ import annotations
 
 from typing import Literal, TypeVar

tests/conftest.py

@@ -5,45 +5,25 @@
 import pytest
 import pytest_asyncio
 
+from sqlalchemy import insert
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 
 from sqlalchemy_crud_plus import CRUDPlus
-from tests.models.basic import Base as BasicBase
-from tests.models.basic import Ins, InsPks
+from tests.models.basic import Base, Ins, InsPks
 from tests.models.relations import RelationBase, RelCategory, RelPost, RelProfile, RelRole, RelUser, user_role
 from tests.schemas.relations import RelPostCreate, RelProfileCreate, RelRoleCreate, RelUserCreate
 
-# Database configuration
 _async_engine = create_async_engine('sqlite+aiosqlite:///:memory:', future=True, echo=False)
-_async_session_factory = async_sessionmaker(_async_engine, autoflush=False, expire_on_commit=False)
+_async_db_session = async_sessionmaker(_async_engine, autoflush=False, expire_on_commit=False)
 
 
-@pytest_asyncio.fixture(scope='function', autouse=True)
-async def setup_database() -> AsyncGenerator[None, None]:
-    """Setup and teardown database for each test function."""
+@pytest_asyncio.fixture(scope='session', autouse=True)
+async def setup_database():
+    """Set up the test database."""
     async with _async_engine.begin() as conn:
-        await conn.run_sync(BasicBase.metadata.create_all)
+        await conn.run_sync(Base.metadata.create_all)
         await conn.run_sync(RelationBase.metadata.create_all)
 
-    yield
-
-    async with _async_engine.begin() as conn:
-        await conn.run_sync(BasicBase.metadata.drop_all)
-        await conn.run_sync(RelationBase.metadata.drop_all)
-
-
-@pytest_asyncio.fixture
-async def db_session() -> AsyncGenerator[AsyncSession, None]:
-    """Provide a database session for testing."""
-    async with _async_session_factory() as session:
-        yield session
-
-
-@pytest_asyncio.fixture
-async def db_session_factory() -> AsyncGenerator[async_sessionmaker[AsyncSession], None]:
-    """Provide a session factory for testing."""
-    yield _async_session_factory
-
 
 @pytest.fixture
 def crud_ins() -> CRUDPlus[Ins]:
@@ -58,166 +38,158 @@ def crud_ins_pks() -> CRUDPlus[InsPks]:
 
 
 @pytest_asyncio.fixture
-async def populated_db(db_session: AsyncSession, crud_ins: CRUDPlus[Ins]) -> list[Ins]:
+async def async_db_session() -> AsyncGenerator[AsyncSession, None]:
+    """Provide a database session for testing."""
+    async with _async_db_session() as session:
+        yield session
+
+
+@pytest_asyncio.fixture
+async def populated_db(async_db_session: AsyncSession, crud_ins: CRUDPlus[Ins]) -> list[Ins]:
     """Provide a database populated with test data."""
-    async with db_session.begin():
+    async with async_db_session.begin():
         test_data = [Ins(name=f'item_{i}', del_flag=(i % 2 == 0)) for i in range(1, 11)]
-        db_session.add_all(test_data)
-        await db_session.flush()
+        async_db_session.add_all(test_data)
         return test_data
 
 
 @pytest_asyncio.fixture
-async def populated_db_pks(db_session: AsyncSession) -> dict[str, list[InsPks]]:
+async def populated_db_pks(async_db_session: AsyncSession) -> dict[str, list[InsPks]]:
     """Provide a database populated with composite key test data."""
-    async with db_session.begin():
+    async with async_db_session.begin():
         men_data = [InsPks(id=i, name=f'man_{i}', sex='men') for i in range(1, 4)]
-        women_data = [InsPks(id=i, name=f'woman_{i}', sex='women') for i in range(4, 7)]
+        women_data = [InsPks(id=i, name=f'woman_{i}', sex='women') for i in range(1, 4)]
         all_data = men_data + women_data
 
-        db_session.add_all(all_data)
-        await db_session.flush()
+        async_db_session.add_all(all_data)
+        await async_db_session.flush()
 
         return {'men': men_data, 'women': women_data, 'all': all_data}
 
 
 @pytest.fixture
-def rel_user_crud() -> CRUDPlus[RelUser]:
+def rel_crud_user() -> CRUDPlus[RelUser]:
     """Provide CRUD instance for RelUser model."""
     return CRUDPlus(RelUser)
 
 
 @pytest.fixture
-def rel_post_crud() -> CRUDPlus[RelPost]:
+def rel_crud_post() -> CRUDPlus[RelPost]:
     """Provide CRUD instance for RelPost model."""
     return CRUDPlus(RelPost)
 
 
 @pytest.fixture
-def rel_profile_crud() -> CRUDPlus[RelProfile]:
+def rel_crud_profile() -> CRUDPlus[RelProfile]:
     """Provide CRUD instance for RelProfile model."""
     return CRUDPlus(RelProfile)
 
 
 @pytest.fixture
-def rel_category_crud() -> CRUDPlus[RelCategory]:
+def rel_crud_category() -> CRUDPlus[RelCategory]:
     """Provide CRUD instance for RelCategory model."""
     return CRUDPlus(RelCategory)
 
 
 @pytest.fixture
-def rel_role_crud() -> CRUDPlus[RelRole]:
+def rel_crud_role() -> CRUDPlus[RelRole]:
     """Provide CRUD instance for RelRole model."""
     return CRUDPlus(RelRole)
 
 
 @pytest_asyncio.fixture
-async def rel_sample_users(db_session: AsyncSession) -> list[RelUser]:
+async def rel_sample_users(async_db_session: AsyncSession) -> list[RelUser]:
     """Create sample relation users."""
-    async with db_session.begin():
+    async with async_db_session.begin():
         users = []
         for i in range(1, 4):
             user_data = RelUserCreate(name=f'user_{i}')
             user = RelUser(**user_data.model_dump())
-            db_session.add(user)
+            async_db_session.add(user)
             users.append(user)
-        await db_session.flush()
         return users
 
 
 @pytest_asyncio.fixture
-async def rel_sample_profiles(db_session: AsyncSession, rel_sample_users: list[RelUser]) -> list[RelProfile]:
+async def rel_sample_profiles(async_db_session: AsyncSession, rel_sample_users: list[RelUser]) -> list[RelProfile]:
     """Create sample relation profiles."""
-    async with db_session.begin():
+    async with async_db_session.begin():
         profiles = []
-        for i, user in enumerate(rel_sample_users[:2]):  # Only first 2 users get profiles
+        for i, user in enumerate(rel_sample_users[:2]):
             profile_data = RelProfileCreate(bio=f'Bio for {user.name}')
             profile = RelProfile(user_id=user.id, **profile_data.model_dump())
-            db_session.add(profile)
+            async_db_session.add(profile)
             profiles.append(profile)
-        await db_session.flush()
         return profiles
 
 
 @pytest_asyncio.fixture
-async def rel_sample_categories(db_session: AsyncSession) -> list[RelCategory]:
+async def rel_sample_categories(async_db_session: AsyncSession) -> list[RelCategory]:
     """Create sample relation categories."""
-    async with db_session.begin():
-        categories = []
-        # Root categories
-        tech = RelCategory(name='Technology')
-        science = RelCategory(name='Science')
-        db_session.add_all([tech, science])
-        await db_session.flush()
-
-        # Sub categories
+    async with async_db_session.begin():
+        tech = RelCategory(name='Technology', parent_id=None)
+        science = RelCategory(name='Science', parent_id=None)
+        async_db_session.add_all([tech, science])
+        await async_db_session.flush()
+
         programming = RelCategory(name='Programming', parent_id=tech.id)
         ai = RelCategory(name='AI', parent_id=tech.id)
-        db_session.add_all([programming, ai])
-        await db_session.flush()
-
+        async_db_session.add_all([programming, ai])
         categories = [tech, science, programming, ai]
         return categories
 
 
 @pytest_asyncio.fixture
 async def rel_sample_posts(
-    db_session: AsyncSession, rel_sample_users: list[RelUser], rel_sample_categories: list[RelCategory]
+    async_db_session: AsyncSession, rel_sample_users: list[RelUser], rel_sample_categories: list[RelCategory]
 ) -> list[RelPost]:
     """Create sample relation posts."""
-    async with db_session.begin():
+    async with async_db_session.begin():
         posts = []
         for i in range(6):
             post_data = RelPostCreate(
                 title=f'Post {i + 1}',
                 category_id=rel_sample_categories[i % len(rel_sample_categories)].id if i < 4 else None,
             )
             post = RelPost(author_id=rel_sample_users[i % len(rel_sample_users)].id, **post_data.model_dump())
-            db_session.add(post)
+            async_db_session.add(post)
             posts.append(post)
-        await db_session.flush()
         return posts
 
 
 @pytest_asyncio.fixture
-async def rel_sample_roles(db_session: AsyncSession) -> list[RelRole]:
+async def rel_sample_roles(async_db_session: AsyncSession) -> list[RelRole]:
     """Create sample relation roles."""
-    async with db_session.begin():
+    async with async_db_session.begin():
         roles = []
-        for role_name in ['admin', 'editor', 'viewer']:
+        for role_name in ['admin', 'editor']:
             role_data = RelRoleCreate(name=role_name)
             role = RelRole(**role_data.model_dump())
-            db_session.add(role)
+            async_db_session.add(role)
             roles.append(role)
-        await db_session.flush()
         return roles
 
 
 @pytest_asyncio.fixture
 async def rel_sample_data(
-    db_session: AsyncSession,
+    async_db_session: AsyncSession,
     rel_sample_users: list[RelUser],
     rel_sample_profiles: list[RelProfile],
     rel_sample_categories: list[RelCategory],
     rel_sample_posts: list[RelPost],
     rel_sample_roles: list[RelRole],
 ) -> dict:
     """Create complete relation sample data with relationships."""
-    async with db_session.begin():
-        # Create user-role relationships
-        await db_session.execute(
-            user_role.insert().values(
+    async with async_db_session.begin():
+        await async_db_session.execute(
+            insert(user_role).values(
                 [
-                    {'user_id': rel_sample_users[0].id, 'role_id': rel_sample_roles[0].id},  # user_1 is admin
-                    {'user_id': rel_sample_users[1].id, 'role_id': rel_sample_roles[1].id},  # user_2 is editor
-                    {'user_id': rel_sample_users[0].id, 'role_id': rel_sample_roles[1].id},  # user_1 is also editor
+                    {'user_id': rel_sample_users[0].id, 'role_id': rel_sample_roles[0].id},
+                    {'user_id': rel_sample_users[1].id, 'role_id': rel_sample_roles[1].id},
                 ]
             )
         )
 
-        await db_session.flush()
-        await db_session.commit()
-
     return {
         'users': rel_sample_users,
         'profiles': rel_sample_profiles,

tests/models/relations.py

@@ -3,14 +3,15 @@
 from __future__ import annotations
 
 from sqlalchemy import Column, ForeignKey, Integer, String, Table
-from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
+from sqlalchemy.orm import DeclarativeBase, Mapped, MappedAsDataclass, declared_attr, mapped_column, relationship
 
 
-class RelationBase(DeclarativeBase):
-    pass
+class RelationBase(MappedAsDataclass, DeclarativeBase):
+    @declared_attr.directive
+    def __tablename__(cls) -> str:
+        return cls.__name__.lower()
 
 
-# Association table for Many-to-Many relationship between User and Role
 user_role = Table(
     'user_role',
     RelationBase.metadata,
@@ -22,67 +23,52 @@ class RelationBase(DeclarativeBase):
 class RelUser(RelationBase):
     __tablename__ = 'rel_user'
 
-    id: Mapped[int] = mapped_column(primary_key=True)
-    name: Mapped[str] = mapped_column(String(50))
+    id: Mapped[int] = mapped_column(init=False, primary_key=True, autoincrement=True)
+    name: Mapped[str] = mapped_column(String(30))
 
-    # One-to-One relationship: User has one profile
-    profile: Mapped[RelProfile | None] = relationship(back_populates='user', uselist=False)
-
-    # One-to-Many relationship: User has many posts
-    posts: Mapped[list[RelPost]] = relationship(back_populates='author')
-
-    # Many-to-Many relationship: User has many roles, Role has many users
-    roles: Mapped[list[RelRole]] = relationship(secondary=user_role, back_populates='users')
+    profile: Mapped[RelProfile | None] = relationship(init=False, back_populates='user', uselist=False)
+    posts: Mapped[list[RelPost]] = relationship(init=False, back_populates='author')
+    roles: Mapped[list[RelRole]] = relationship(init=False, secondary=user_role, back_populates='users')
 
 
 class RelProfile(RelationBase):
     __tablename__ = 'rel_profile'
 
-    id: Mapped[int] = mapped_column(primary_key=True)
+    id: Mapped[int] = mapped_column(init=False, primary_key=True, autoincrement=True)
     user_id: Mapped[int] = mapped_column(ForeignKey('rel_user.id'), unique=True)
-    bio: Mapped[str] = mapped_column(String(200))
+    bio: Mapped[str] = mapped_column(String(100))
 
-    # One-to-One relationship: Profile belongs to one user
-    user: Mapped[RelUser] = relationship(back_populates='profile')
+    user: Mapped[RelUser] = relationship(init=False, back_populates='profile')
 
 
 class RelCategory(RelationBase):
     __tablename__ = 'rel_category'
 
-    id: Mapped[int] = mapped_column(primary_key=True)
-    name: Mapped[str] = mapped_column(String(50))
+    id: Mapped[int] = mapped_column(init=False, primary_key=True, autoincrement=True)
+    name: Mapped[str] = mapped_column(String(30))
     parent_id: Mapped[int | None] = mapped_column(ForeignKey('rel_category.id'))
 
-    # Self-referential relationship: Category belongs to parent category
-    parent: Mapped[RelCategory | None] = relationship('RelCategory', remote_side=[id], back_populates='children')
-
-    # Self-referential relationship: Category has many child categories
-    children: Mapped[list[RelCategory]] = relationship('RelCategory', back_populates='parent')
-
-    # One-to-Many relationship: Category has many posts
-    posts: Mapped[list[RelPost]] = relationship(back_populates='category')
+    parent: Mapped[RelCategory | None] = relationship(init=False, remote_side=[id], back_populates='children')
+    children: Mapped[list[RelCategory]] = relationship(init=False, back_populates='parent')
+    posts: Mapped[list[RelPost]] = relationship(init=False, back_populates='category')
 
 
 class RelPost(RelationBase):
     __tablename__ = 'rel_post'
 
-    id: Mapped[int] = mapped_column(primary_key=True)
-    title: Mapped[str] = mapped_column(String(100))
+    id: Mapped[int] = mapped_column(init=False, primary_key=True, autoincrement=True)
+    title: Mapped[str] = mapped_column(String(50))
     author_id: Mapped[int] = mapped_column(ForeignKey('rel_user.id'))
     category_id: Mapped[int | None] = mapped_column(ForeignKey('rel_category.id'))
 
-    # Many-to-One relationship: Post belongs to one author (user)
-    author: Mapped[RelUser] = relationship(back_populates='posts')
-
-    # Many-to-One relationship: Post belongs to one category (optional)
-    category: Mapped[RelCategory | None] = relationship(back_populates='posts')
+    author: Mapped[RelUser] = relationship(init=False, back_populates='posts')
+    category: Mapped[RelCategory | None] = relationship(init=False, back_populates='posts')
 
 
 class RelRole(RelationBase):
     __tablename__ = 'rel_role'
 
-    id: Mapped[int] = mapped_column(primary_key=True)
-    name: Mapped[str] = mapped_column(String(50))
+    id: Mapped[int] = mapped_column(init=False, primary_key=True, autoincrement=True)
+    name: Mapped[str] = mapped_column(String(20))
 
-    # Many-to-Many relationship: Role has many users, User has many roles
-    users: Mapped[list[RelUser]] = relationship(secondary=user_role, back_populates='roles')
+    users: Mapped[list[RelUser]] = relationship(init=False, secondary=user_role, back_populates='roles')