$ alembic revision --autogenerate -m "YOUR_COMMENT_HERE"

alembic upgrade head

alembic upgrade +1

alembic downgrade base

alembic downgrade -1

router = APIRouter(
    prefix="/sample",
    tags=["sample"],
    dependencies=[Depends(get_first_user), Dpends(get_second_user)],
)

async def update(
    body: SampleRequest,
    sample_id: PydanticObjectId = Path(...),
    sample_service: SampleService = Depends(SampleService),
)

def get_user():
    # ユーザーを取得する処理
    pass

def get_admin():
    # 管理者を取得する処理
    pass

app.dependency_overrides[get_user] = lambda: "mocked user"

class UserService:
    pass

app.dependency_overrides[UserService] = lambda: "mocked service"

def can_perform(permission: str):
    async def verify_permission():
        pass
    return verify_permission

app.dependency_overrides[can_perform('read')] = lambda: "mocked permission check"

from dotenv import dotenv_values
from sqlalchemy import create_engine
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker, declarative_base

DB_URL = "mysql+pymysql://%s:%s@%s/%s?charset=utf8" % (
	dotenv_values(".env")["MYSQL_USER"],
	dotenv_values(".env")["MYSQL_PASSWORD"],
	dotenv_values(".env")["MYSQL_HOST"],
	dotenv_values(".env")["MYSQL_DATABASE"],
)

ASYNC_DB_URL = "mysql+aiomysql://%s:%s@%s/%s?charset=utf8" % (
	dotenv_values(".env")["MYSQL_USER"],
	dotenv_values(".env")["MYSQL_PASSWORD"],
	dotenv_values(".env")["MYSQL_HOST"],
	dotenv_values(".env")["MYSQL_DATABASE"],
)

engine = create_engine(DB_URL, echo=True)
async_engine = create_async_engine(ASYNC_DB_URL, echo=True)

session = sessionmaker(
	autocommit=False,
	autoflush=False,
	bind=engine
)

async_session = sessionmaker(
	autocommit=False,
	autoflush=False,
	bind=async_engine,
	class_=AsyncSession
)

Base = declarative_base()


async def get_db():
	async with async_session() as s:
		yield s


def get_sync_db():
	with session() as s:
		yield s

from typing import TypeVar, Generic

from pydantic.generics import GenericModel

T = TypeVar("T")


class LanguageDict(GenericModel, Generic[T]):
	en: T
	ja: T

        @pytest.mark.asyncio
	async def test_should_return_none_if_document_not_found(self):
		result = await self.repo.get_by_id("123456789012345678901234") #型なし、バリデーション満たす文字列
		assert result is None

	@pytest.mark.asyncio
	async def test_should_return_none_if_document_not_found(self):
		result = await self.repo.get_by_id("1234567890")
		assert result is None

	@pytest.mark.asyncio
	async def test_should_return_none_if_document_not_found(self):
		dummy_object_id: PydanticObjectId = "1234567890"
		result = await self.repo.get_by_id(dummy_object_id)
		assert result is None

	@pytest.mark.asyncio
	async def test_should_return_none_if_document_not_found(self):
		dummy_object_id = PydanticObjectId("1234567890")
		result = await self.repo.get_by_id(dummy_object_id)
		assert result is None

@router.get('/', response_model=list[Stimulus])
async def index(
		stimulus_service: StimulusService = Depends(StimulusService),
		query: BasePaginateQuery = Depends()) -> list[Stimulus]:
	return await stimulus_service.get(paginate=query.dict())

class BasePaginateQuery(BaseModel):
	page: int = Query(1, gt=0)
	limit: int = Query(10, gt=0, le=100)

from pydantic import BaseModel

class Sample(BaseModel):
    name: str

    class Config:
        extra = "forbid"

# Now using an unknown key will cause a ValidationError
Sample.parse_obj({'name': 'Test', 'unknown_key': 'value'})  

path: constr(regex=r"^\w+(\.\w+)*$") = Field(..., min_length=1)

@router.get('/sample/{param}')

@router.get('/{param}')

curl http://localhost/sample

class Sample(BaseModel):
  query1: Optional[str] = None
  query2: List[str] = [] # Queryで書いても同じ

@router.get('/')
async def get(
  tags: List[str] = Query([], alias="tags")
):

class SignupRequestProviderEmail(BaseModel):
	type: Literal['email'] = 'email'
	email: EmailStr
	password: SecretStr

	@field_validator('password')
	def has_min_length(cls, password: SecretStr) -> SecretStr:
		min_length = 8
		if len(password) < min_length:
			raise ValueError(f"ensure this value has at least {min_length} characters")
		return password

	@field_validator('password')
	def has_max_length(cls, password: SecretStr) -> SecretStr:
		max_length = 32
		if len(password) > max_length:
			raise ValueError(f"ensure this value has at most {max_length} characters")
		return password

	@field_validator('password')
	def matches_regex(cls, password: SecretStr) -> SecretStr:
		regex = r"^(?=.*[a-zA-Z])(?=.*\d)(?=.*[!@#$%^&*]).+$"
		if not re.match(regex, password.get_secret_value()):
			raise ValueError("Password must have at least one letter, one number and one special character.")
		return password

import importlib
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context
from sqlmodel import SQLModel
from dotenv import dotenv_values
import os

# Load environment variables from .env file
config = dotenv_values(".env")

# Retrieve database connection info from environment variables
DB_USER = config.get("MYSQL_USER")
DB_PASSWORD = config.get("MYSQL_PASSWORD")
DB_HOST = config.get("MYSQL_HOST")
DB_NAME = config.get("MYSQL_DATABASE")

# Construct the database URL
DATABASE_URL = f"mysql+pymysql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}/{DB_NAME}?charset=utf8"

# Update the Alembic config with the database URL
config = context.config
config.set_main_option('sqlalchemy.url', DATABASE_URL)

# Interpret the config file for Python logging
if config.config_file_name is not None:
	fileConfig(config.config_file_name)


# Function to dynamically import all models from the specified root directory
# This function retrieves all Python files in the specified directory and files which end with '_model.py'
def import_all_models(root_directory):
	for dirpath, _, filenames in os.walk(root_directory):
		for filename in [f for f in filenames if f.endswith('.py') and f != '__init__.py']:
			if filename.endswith('_model.py'):
				module_name = filename[:-3]  # Remove .py extension
				module_path = os.path.join(dirpath, module_name).replace(os.sep, '.')
				importlib.import_module(module_path)


# Import all models in the specified root directory
import_all_models('src/module')

# Add your model's MetaData object here
# for 'autogenerate' support
target_metadata = SQLModel.metadata


def run_migrations_offline() -> None:
	"""Run migrations in 'offline' mode.

	This configures the context with just a URL
	and not an Engine, though an Engine is acceptable
	here as well. By skipping the Engine creation
	we don't even need a DBAPI to be available.

	Calls to context.execute() here emit the given string to the
	script output.
	"""
	url = config.get_main_option("sqlalchemy.url")
	context.configure(
		url=url, target_metadata=target_metadata, literal_binds=True
	)

	with context.begin_transaction():
		context.run_migrations()


def run_migrations_online() -> None:
	"""Run migrations in 'online' mode.

	In this scenario we need to create an Engine
	and associate a connection with the context.
	"""
	connectable = engine_from_config(
		config.get_section(config.config_ini_section),
		prefix="sqlalchemy.",
		poolclass=pool.NullPool,
	)

	with connectable.connect() as connection:
		context.configure(
			connection=connection, target_metadata=target_metadata
		)

		with context.begin_transaction():
			context.run_migrations()


if context.is_offline_mode():
	run_migrations_offline()
else:
	run_migrations_online()

import asyncio
import importlib
import os
from logging.config import fileConfig

from dotenv import dotenv_values
from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import async_engine_from_config

from alembic import context
from sqlmodel import SQLModel

# Load environment variables from .env file
config = dotenv_values(".env")

# Retrieve database connection info from environment variables
DB_USER = config.get("MYSQL_USER")
DB_PASSWORD = config.get("MYSQL_PASSWORD")
DB_HOST = config.get("MYSQL_HOST")
DB_NAME = config.get("MYSQL_DATABASE")

# Construct the database URL
DATABASE_URL = f"mysql+aiomysql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}/{DB_NAME}?charset=utf8"

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config

# Update the Alembic config with the database URL
config.set_main_option('sqlalchemy.url', DATABASE_URL)

# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
	fileConfig(config.config_file_name)


# Function to dynamically import all models from the specified root directory
# This function retrieves all Python files in the specified directory and files which end with '_model.py'
def import_all_models(root_directory):
	for dirpath, _, filenames in os.walk(root_directory):
		for filename in [f for f in filenames if f.endswith('.py') and f != '__init__.py']:
			if filename.endswith('_model.py'):
				module_name = filename[:-3]  # Remove .py extension
				module_path = os.path.join(dirpath, module_name).replace(os.sep, '.')
				importlib.import_module(module_path)


# Import all models in the specified root directory
import_all_models('src/module')

# add your model's MetaData object here
# for 'autogenerate' support
target_metadata = SQLModel.metadata


# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.


def run_migrations_offline() -> None:
	"""Run migrations in 'offline' mode.

	This configures the context with just a URL
	and not an Engine, though an Engine is acceptable
	here as well.  By skipping the Engine creation
	we don't even need a DBAPI to be available.

	Calls to context.execute() here emit the given string to the
	script output.

	"""
	url = config.get_main_option("sqlalchemy.url")
	context.configure(
		url=url,
		target_metadata=target_metadata,
		literal_binds=True,
		dialect_opts={"paramstyle": "named"},
	)

	with context.begin_transaction():
		context.run_migrations()


def do_run_migrations(connection: Connection) -> None:
	context.configure(connection=connection, target_metadata=target_metadata)

	with context.begin_transaction():
		context.run_migrations()


async def run_async_migrations() -> None:
	"""In this scenario we need to create an Engine
	and associate a connection with the context.

	"""

	connectable = async_engine_from_config(
		config.get_section(config.config_ini_section, {}),
		prefix="sqlalchemy.",
		poolclass=pool.NullPool,
	)

	async with connectable.connect() as connection:
		await connection.run_sync(do_run_migrations)

	await connectable.dispose()


def run_migrations_online() -> None:
	"""Run migrations in 'online' mode."""

	asyncio.run(run_async_migrations())


if context.is_offline_mode():
	run_migrations_offline()
else:
	run_migrations_online()

"""${message}

Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa
# NOTE: This import is required for the Alembic autogeneration to work when use SQLModel
import sqlmodel
${imports if imports else ""}

# revision identifiers, used by Alembic.
revision: str = ${repr(up_revision)}
down_revision: Union[str, None] = ${repr(down_revision)}
branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}


def upgrade() -> None:
    ${upgrades if upgrades else "pass"}


def downgrade() -> None:
    ${downgrades if downgrades else "pass"}

language: AvailableLanguageEnum | None = Field(sa_type=AutoString)

class UserModel(TimestampModel, SQLModel, table=True):
	__tablename__ = "user"
	id: int | None = Field(primary_key=True, default=None)
	name: str
	language: AvailableLanguageEnum | None = Field(sa_type=AutoString)
	timezone: TimeZoneEnum | None = Field(sa_type=AutoString)
	icon_url: Optional[str] | None

	auth_provider_email: Optional["UserAuthProviderEmailModel"] = Relationship(
		back_populates="user",
		sa_relationship_kwargs={"cascade": "all, delete-orphan"}
	)

class UserAuthProviderEmailModel(SQLModel, table=True):
	__tablename__ = "user_auth_provider_email"
	user_id: int = Field(
		sa_column=Column(
			"user_id",
			ForeignKey("user.id", ondelete="CASCADE"),
			primary_key=True,
		),
	)
	email: str = Field(index=True, unique=True)
	password: str
	verified: bool = False

	user: "UserModel" = Relationship(back_populates="auth_provider_email")

class Example(SQLModel, table=True):
    __table_args__ = (UniqueConstraint('workspace_id', 'user_id', name='unique_workspace_member'),)
    workspace_id: int
    user_id: int

class ContactModel(TimestampModel, SQLModel, table=True):
	__tablename__ = "contact"
	__table_args__ = (
		Index(
			'idx_contact_fulltext', # index name
			'first_name', # colmuns 
			'last_name',
			mysql_prefix='FULLTEXT',
			mysql_with_parser='ngram'
		),
	)

		for sort in query.sorts:
			sort_col = getattr(ContactModel, sort.field, None)
			if sort_col is None:
				raise RuntimeError(f"Invalid sort field: {sort.field} does not exist in ContactModel")
			statement = statement.order_by(sort_col.desc() if sort.order == 'desc' else sort_col.asc())

statement.subquery()

count_statement = select(func.count()).select_from(statement.subquery())
contact_count = await self.db_session.exec(count_statement)
total_count = contact_count.first()  # カウント結果

E           sqlalchemy.exc.InvalidRequestError: When initializing mapper Mapper[WorkspaceModel(workspace)], expression 'WorkspaceMemberModel' failed to locate a name ('WorkspaceMemberModel'). If this is a class name, consider adding this relationship() to the <class 'src.module.workspace.infrastructure.orm.workspace_model.WorkspaceModel'> class after both dependent classes have been defined.

def check_object_state(obj):
	# InstanceStateオブジェクトを取得
	state = obj._sa_instance_state

	# セッションが存在するかを確認
	session = state.session

	# デタッチされているかを確認
	detached = state.detached

	print(f"Session: {session}")
	print(f"Detached: {detached}")

async def get_db_session():
	async with AsyncSession(engine, expire_on_commit=False) as session:
		yield session