I created a function that checks the data from environment variables about the first owner and creates it if it doesn’t exist. If I use this function in a route, it works correctly, but if I add it to the lifespan function, I get an error.
first owner create func
<code>async def first_owner_create(db_session: AsyncSession):
query = select(User).where(
(User.username == settings.FIRST_OWNER_USERNAME)
& (User.email == settings.FIRST_OWNER_EMAIL)
)
result = await db_session.execute(query)
user = result.scalars().first()
if not user:
hashed_password = get_password_hash(settings.FIRST_OWNER_PASSWORD)
user_in = User(
username=settings.FIRST_OWNER_USERNAME,
email=settings.FIRST_OWNER_EMAIL,
password=hashed_password,
role=UserRole.OWNER,
)
db_session.add(user_in)
await db_session.commit()
await db_session.refresh(user_in)
</code>
<code>async def first_owner_create(db_session: AsyncSession):
query = select(User).where(
(User.username == settings.FIRST_OWNER_USERNAME)
& (User.email == settings.FIRST_OWNER_EMAIL)
)
result = await db_session.execute(query)
user = result.scalars().first()
if not user:
hashed_password = get_password_hash(settings.FIRST_OWNER_PASSWORD)
user_in = User(
username=settings.FIRST_OWNER_USERNAME,
email=settings.FIRST_OWNER_EMAIL,
password=hashed_password,
role=UserRole.OWNER,
)
db_session.add(user_in)
await db_session.commit()
await db_session.refresh(user_in)
</code>
async def first_owner_create(db_session: AsyncSession):
query = select(User).where(
(User.username == settings.FIRST_OWNER_USERNAME)
& (User.email == settings.FIRST_OWNER_EMAIL)
)
result = await db_session.execute(query)
user = result.scalars().first()
if not user:
hashed_password = get_password_hash(settings.FIRST_OWNER_PASSWORD)
user_in = User(
username=settings.FIRST_OWNER_USERNAME,
email=settings.FIRST_OWNER_EMAIL,
password=hashed_password,
role=UserRole.OWNER,
)
db_session.add(user_in)
await db_session.commit()
await db_session.refresh(user_in)
lifespan
<code>@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
# Startup
await first_owner_create(SessionDep)
redis = aioredis.from_url("redis://localhost")
FastAPICache.init(RedisBackend(redis), prefix="fastapi-cache")
yield
# Shutdown
</code>
<code>@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
# Startup
await first_owner_create(SessionDep)
redis = aioredis.from_url("redis://localhost")
FastAPICache.init(RedisBackend(redis), prefix="fastapi-cache")
yield
# Shutdown
</code>
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
# Startup
await first_owner_create(SessionDep)
redis = aioredis.from_url("redis://localhost")
FastAPICache.init(RedisBackend(redis), prefix="fastapi-cache")
yield
# Shutdown
Error:
<code>ERROR: Traceback (most recent call last):
File "/home/username/Dev/screen-scout/.venv/lib/python3.12/site-packages/starlette/routing.py", line 732, in lifespan
async with self.lifespan_context(app) as maybe_state:
File "/home/username/.pyenv/versions/3.12.4/lib/python3.12/contextlib.py", line 210, in __aenter__
return await anext(self.gen)
^^^^^^^^^^^^^^^^^^^^^
File "/home/username/Dev/screen-scout/screenscout/main.py", line 18, in lifespan
await first_owner_create(SessionDep)
File "/home/username/Dev/screen-scout/screenscout/auth/service.py", line 128, in first_owner_create
result = await db_session.execute(query)
^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: AsyncSession.execute() missing 1 required positional argument: 'statement'
ERROR: Application startup failed. Exiting.
</code>
<code>ERROR: Traceback (most recent call last):
File "/home/username/Dev/screen-scout/.venv/lib/python3.12/site-packages/starlette/routing.py", line 732, in lifespan
async with self.lifespan_context(app) as maybe_state:
File "/home/username/.pyenv/versions/3.12.4/lib/python3.12/contextlib.py", line 210, in __aenter__
return await anext(self.gen)
^^^^^^^^^^^^^^^^^^^^^
File "/home/username/Dev/screen-scout/screenscout/main.py", line 18, in lifespan
await first_owner_create(SessionDep)
File "/home/username/Dev/screen-scout/screenscout/auth/service.py", line 128, in first_owner_create
result = await db_session.execute(query)
^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: AsyncSession.execute() missing 1 required positional argument: 'statement'
ERROR: Application startup failed. Exiting.
</code>
ERROR: Traceback (most recent call last):
File "/home/username/Dev/screen-scout/.venv/lib/python3.12/site-packages/starlette/routing.py", line 732, in lifespan
async with self.lifespan_context(app) as maybe_state:
File "/home/username/.pyenv/versions/3.12.4/lib/python3.12/contextlib.py", line 210, in __aenter__
return await anext(self.gen)
^^^^^^^^^^^^^^^^^^^^^
File "/home/username/Dev/screen-scout/screenscout/main.py", line 18, in lifespan
await first_owner_create(SessionDep)
File "/home/username/Dev/screen-scout/screenscout/auth/service.py", line 128, in first_owner_create
result = await db_session.execute(query)
^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: AsyncSession.execute() missing 1 required positional argument: 'statement'
ERROR: Application startup failed. Exiting.
I created a route that triggers the creation of the first owner when called, and it works correctly. However, I need this to happen when the application starts. I added it to the startup event, but I get an error related to a missing statement argument, even though it is present.
1