Menü
Modern async ORM with type-safe models and efficient queries
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Basierend auf der SOC-Berufsklassifikation
Manage Bitbucket pull requests, repositories, and pipelines and Jira issues from the terminal with the bj CLI. bj mirrors the GitHub gh CLI, so gh pr/issue/repo commands map to bj pr/issue/repo, no MCP server or API wiring needed. Installs the bj CLI automatically if it is missing. Use when working with Bitbucket or Jira, opening or reviewing pull requests, managing Jira issues, running pipelines, or when a git branch name contains a Jira issue key like PROJ-42.
Use when auditing a paid ad account for incremental contribution, wasted spend, or measurement integrity before scaling; runs a typed 20-item ROAS profile with verified vetoes and a SHIP/FIX/BLOCK/UNDECIDED gate on own exported data. Not for campaign structure design — use campaign-architect; not for creative production — use ad-creative-builder. 付费广告账户审计/ROAS评分
Use when the user asks to "write ad copy", "generate RSA headlines", or "build ad creative at volume"; produces ad units — RSA headlines/descriptions, hooks, and an angle matrix — message-matched to the destination landing page. Not for scoring an ad account — use ad-account-auditor; not for the post-click page — use landing-optimizer; not for organic articles — use content-writer. 广告创意/广告文案/RSA标题
Use when the user asks to "design an A/B test", "set up a creative/landing test", "run an incrementality test", or "is this result statistically and practically material?"; produces a hypothesis, variant matrix, sample-size/duration/power plan, and a documented effect/uncertainty read from own exported results. It applies only a precommitted owner-approved action rule; the statistical helper never chooses a business action. Not for producing variants — use ad-creative-builder; not for reading back one shipped change — use paid-measurement-loop. 广告AB测试设计/实验设计/显著性判定/增效测试
Use when platform-reported conversions disagree with GA4/ecommerce, when you suspect Meta and Google are double-counting the same sales, or for a standing (monthly) reconciliation workbook that de-dups stacked credit against an order-ID truth set, normalizes attribution windows and currency, compares attribution models, and reads incrementality from a geo/holdout test. Not for the point-in-time R2 veto or RQS gate — use ad-account-auditor; not for the ROI/ROAS ratio math itself — use roi-calculator; not for organic dark-social share attribution or GA4 direct-traffic decomposition — use dark-social-attributor. 付费广告归因对账/去重/增量
Use when the user asks to "analyze my target audience", "build an audience profile for influencer targeting", "research a niche community", or "deep-dive a subculture before partnering with creators"; in audience mode produces demographic/psychographic profiles, a platform-priority matrix, named personas, and an influencer-selection criteria set, and in niche mode produces a community map, culture decode (language/norms/taboos), key-voice tiers, a Brand Fit Score, and a phased entry strategy. Not for finding specific creators to contract — use influencer-discovery; not for scoring a shortlist on ACE — use fit-scorer. 目标受众画像/人群分析 · 细分社群/亚文化调研
| name | sqlalchemy-2-0 |
| description | Modern async ORM with type-safe models and efficient queries |
| when_to_use | Building database backends, APIs, data services with async support |
from sqlalchemy.ext.asyncio import AsyncAttrs, async_sessionmaker, create_async_engine, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
import asyncio
# Base class for models
class Base(AsyncAttrs, DeclarativeBase):
pass
# Async engine
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
# Session factory
async_session = async_sessionmaker(engine, expire_on_commit=False)
# Example model
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50))
email: Mapped[str] = mapped_column(String(100))
async def create_user(name: str, email: str) -> User:
async with async_session() as session:
async with session.begin():
user = User(name=name, email=email)
session.add(user)
await session.flush() # Get the ID
return user
async def get_user(user_id: int) -> User | None:
async with async_session() as session:
result = await session.execute(select(User).where(User.id == user_id))
return result.scalar_one_or_none()
async def update_user_email(user_id: int, new_email: str) -> bool:
async with async_session() as session:
result = await session.execute(
update(User).where(User.id == user_id).values(email=new_email)
)
await session.commit()
return result.rowcount > 0
from typing_extensions import Annotated
from typing import List, Optional
# Reusable column types
intpk = Annotated[int, mapped_column(primary_key=True)]
str50 = Annotated[str, mapped_column(String(50))]
created_at = Annotated[datetime, mapped_column(insert_default=func.now())]
class Post(Base):
__tablename__ = "posts"
id: Mapped[intpk]
title: Mapped[str50]
content: Mapped[str] = mapped_column(Text)
author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
created: Mapped[created_at]
# Relationships
author: Mapped["User"] = relationship(back_populates="posts")
tags: Mapped[List["Tag"]] = relationship(secondary="post_tags")
class Post(Base):
__tablename__ = "posts"
id = mapped_column(Integer, primary_key=True)
title = mapped_column(String(50))
content = mapped_column(Text)
author_id = mapped_column(ForeignKey("users.id"))
author = relationship("User", back_populates="posts")
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
posts: Mapped[List["Post"]] = relationship(
back_populates="author",
cascade="all, delete-orphan"
)
class Post(Base):
__tablename__ = "posts"
id: Mapped[int] = mapped_column(primary_key=True)
author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
author: Mapped["User"] = relationship(back_populates="posts")
association_table = Table(
"post_tags",
Base.metadata,
Column("post_id", ForeignKey("posts.id"), primary_key=True),
Column("tag_id", ForeignKey("tags.id"), primary_key=True)
)
class Post(Base):
__tablename__ = "posts"
id: Mapped[int] = mapped_column(primary_key=True)
tags: Mapped[List["Tag"]] = relationship(
secondary=association_table,
back_populates="posts"
)
class Tag(Base):
__tablename__ = "tags"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50), unique=True)
posts: Mapped[List["Post"]] = relationship(
secondary=association_table,
back_populates="tags"
)
from sqlalchemy import select, and_, or_
# Get all users
async def get_all_users():
async with async_session() as session:
result = await session.execute(select(User))
return result.scalars().all()
# Filter with conditions
async def get_users_by_name(name: str):
async with async_session() as session:
stmt = select(User).where(User.name.ilike(f"%{name}%"))
result = await session.execute(stmt)
return result.scalars().all()
# Complex conditions
async def search_users(name: str = None, email: str = None):
async with async_session() as session:
conditions = []
if name:
conditions.append(User.name.ilike(f"%{name}%"))
if email:
conditions.append(User.email.ilike(f"%{email}%"))
if conditions:
stmt = select(User).where(and_(*conditions))
else:
stmt = select(User)
result = await session.execute(stmt)
return result.scalars().all()
from sqlalchemy.orm import selectinload, joinedload
# Eager load relationships
async def get_posts_with_author():
async with async_session() as session:
stmt = select(Post).options(selectinload(Post.author))
result = await session.execute(stmt)
return result.scalars().all()
# Joined loading for single relationships
async def get_post_with_tags(post_id: int):
async with async_session() as session:
stmt = select(Post).options(
joinedload(Post.author),
selectinload(Post.tags)
).where(Post.id == post_id)
result = await session.execute(stmt)
return result.scalar_one_or_none()
async def get_posts_paginated(page: int, size: int):
async with async_session() as session:
offset = (page - 1) * size
stmt = select(Post).offset(offset).limit(size).order_by(Post.created.desc())
result = await session.execute(stmt)
return result.scalars().all()
from sqlalchemy import func
async def get_user_post_count():
async with async_session() as session:
stmt = (
select(User.name, func.count(Post.id).label("post_count"))
.join(Post)
.group_by(User.id, User.name)
.order_by(func.count(Post.id).desc())
)
result = await session.execute(stmt)
return result.all()
async def create_post(title: str, content: str, author_id: int):
async with async_session() as session:
async with session.begin():
post = Post(title=title, content=content, author_id=author_id)
session.add(post)
return post
from fastapi import Depends
async def get_db_session():
async with async_session() as session:
try:
yield session
finally:
await session.close()
async def create_user_endpoint(
user_data: UserCreate,
session: AsyncSession = Depends(get_db_session)
):
user = User(**user_data.dict())
session.add(user)
await session.commit()
await session.refresh(user)
return user
from sqlalchemy.ext.asyncio import async_scoped_session
import asyncio
# Create scoped session
async_session_scope = async_scoped_session(
async_sessionmaker(engine, expire_on_commit=False),
scopefunc=asyncio.current_task
)
# Use in application
async def some_function():
session = async_session_scope()
# Use session normally
await session.commit()
from sqlalchemy.orm import WriteOnlyMapped
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
posts: WriteOnlyMapped["Post"] = relationship()
async def get_user_posts(user_id: int):
async with async_session() as session:
user = await session.get(User, user_id)
if user:
# Explicit select for collection
stmt = select(Post).where(Post.author_id == user_id)
result = await session.execute(stmt)
return result.scalars().all()
return []
class AsyncSessionWithDefaults(AsyncSession):
async def execute_with_defaults(self, statement, **kwargs):
# Add default options
return await self.execute(statement, **kwargs)
# Use custom session
async_session = async_sessionmaker(
engine,
class_=AsyncSessionWithDefaults,
expire_on_commit=False
)
class RoutingSession(Session):
def get_bind(self, mapper=None, clause=None, **kw):
if mapper and issubclass(mapper.class_, ReadOnlyModel):
return read_engine
return write_engine
class AsyncRoutingSession(AsyncSession):
sync_session_class = RoutingSession
from sqlalchemy import text
async def run_raw_sql():
async with async_session() as session:
result = await session.execute(text("SELECT COUNT(*) FROM users"))
count = result.scalar()
return count
async def run_parameterized_query(user_id: int):
async with async_session() as session:
stmt = text("SELECT * FROM posts WHERE author_id = :user_id")
result = await session.execute(stmt, {"user_id": user_id})
return result.fetchall()
add_all() for bulk insertsstream()# Streaming large results
async def process_all_users():
async with async_session() as session:
result = await session.stream(select(User))
async for user in result.scalars():
# Process user without loading all into memory
await process_user(user)
uv add sqlalchemy[asyncio] # Core SQLAlchemy
uv add asyncpg # PostgreSQL async driver
# or
uv add aiosqlite # SQLite async driver
# or
uv add aiomysql # MySQL async driver
postgresql+asyncpg://user:pass@localhost/dbsqlite+aiosqlite:///database.dbmysql+aiomysql://user:pass@localhost/dbUse Alembic for database migrations:
# Generate migration
uv run alembic revision --autogenerate -m "Add users table"
# Apply migrations
uv run alembic upgrade head