#🔒 followup post - check body for link

108 messages · Page 1 of 1 (latest)

haughty glen
swift depotBOT
#

@haughty glen

Python help channel opened

Remember to:

  • Ask your Python question, not if you can ask or if there's an expert who can help.
  • Show a code sample as text (rather than a screenshot) and the error message, if you've got one.
  • Explain what you expect to happen and what actually happens.

:warning: Do not pip install anything that isn't related to your question, especially if asked to over DMs.

Closes after a period of inactivity, or when you send !close.
haughty glen
#

[fastapi] followup post - check body for link

#

@whole lark what is and how do you use anyio.create_memory_object_stream?

whole lark
haughty glen
whole lark
#

Yeah I saw you tagged me in the commit

haughty glen
#

what

#

you can see that??

whole lark
swift depotBOT
#

backend/main.py line 17

async with create_pool("entries/index.sql") as pool:```
whole lark
haughty glen
haughty glen
whole lark
#

?

#

Probably create the TaskGroup inside the cmgr for create_pool

haughty glen
whole lark
#

cmgr is short for context manager

haughty glen
#

create_pool(path, cmgr = ...)?

whole lark
#

No I mean inside the with block

haughty glen
whole lark
#

Yeah

#

Using the start_soon method of the TaskGroup from anyio.create_task_group

haughty glen
whole lark
#

You pass an async def to run in the background

#

You also need to call async with on the create_task_group

haughty glen
#

wait why does this have to be in the pool again?

#

forgive me for asking

whole lark
#

Will if you cancel your TaskGroup you don't want to cancel your pool cleanup

haughty glen
#

ok got it

# 
@asynccontextmanager
async def lifespan(app: FastAPI):
    async with create_pool("entries/index.sql") as pool:
        async with anyio.create_task_group() as group:
            group.start_soon(delete_in_background())

        yield {"pool": pool}
#

the delete_in_background is the async def i wanna run in the background

whole lark
#

You pass an async function, not that weird object that async functions return

haughty glen
whole lark
#

Yeah not that, you should never create a coroutine without immediately awaiting it

haughty glen
#

alright

#

gotcha

whole lark
#

But in this case we want concurrency so you pass the function

haughty glen
whole lark
#

And the TaskGroup will call the function and await it

haughty glen
whole lark
#

So pass the second half of a memory object stream into your function and async for it

haughty glen
#

i didnt get any of that

#

"second half of a memory object stream" what

whole lark
#

Start with creating a memory object stream

haughty glen
#

ok

#

how big should the buffer be?

whole lark
# 
send, receive = anyio.create_memory_object_stream[None]()
with send, receive:
    ...
#

Smallest possible

haughty glen
#

alright

whole lark
#

I think the default is the smallest possible

haughty glen
haughty glen
whole lark
#

Show what you have

haughty glen
# whole lark Show what you have 
@asynccontextmanager
async def lifespan(app: FastAPI):
    async with create_pool("entries/index.sql") as pool:
        async with create_task_group() as group:
            sender, receiver = create_memory_object_stream[None]()

            with sender, receiver:
                ... # What goes in here?

            group.start_soon(delete_in_background, pool)

        yield {"pool": pool}
whole lark
#

You need to pass your receive end to your delete_in_background

#

You can use functools.partial for kwargs

#

And you'll want to yield your send end

#

And whenever you call send_nowait(None) on the send end it will wake up the task

#

Or it will raise an exception if it's already in the process of waking

#

You'll just need to catch this

haughty glen
#

ok back

haughty glen
haughty glen
# whole lark And you'll want to yield your send end

then this? ```py
@asynccontextmanager
async def lifespan(app: FastAPI):
async with create_pool("entries/index.sql") as pool:
async with create_task_group() as group:
sender, receiver = create_memory_object_streamNone

        with sender, receiver:
            group.start_soon(delete_in_background, pool, receiver)
            yield {"pool": pool, "sender": sender}
whole lark
#

I can't work out if you have the right things indented under the correct things, but that looks close enough

#

I'm on mobile and it messes with pastes

haughty glen
whole lark
#

Does it work?

haughty glen
#

here's the snippet btw

image.png
haughty glen
whole lark
#

Oh right well do that

haughty glen
# whole lark Oh right well do that

how does it look? ```py
async def delete_in_background(pool: Pool, receiver: MemoryObjectReceiveStream[None]):
async with pool.acquire() as conn:
# Get when the next paste will expire
req = await conn.execute(
"""
SELECT id FROM expiries
ORDER BY expiry_timestamp
LIMIT 1
"""
)
row = await req.fetchone()

if not row:
    # No more records to listen out for
    receiver.close()

await sleep_until(row["delete_at"])

async with pool.acquire() as conn:
    await conn.execute("DELETE FROM expiries WHERE id = ?", row["id"])
#

sleep_until is a custom function i made that gets the duration in seconds between two timestamps and sends that. it works with both datetimes and integer timestamps

# 
@overload
async def sleep_until(datetime: dt, /) -> None:
    "Sleep until a given `datetime`."

@overload
async def sleep_until(timestamp: int, /) -> None:
    "Sleep until a given timestamp."

async def sleep_until(_dt_or_ts: dt | int, /) -> None:
    if isinstance(_dt_or_ts, dt):
        time_asleep = _dt_or_ts.timestamp() - dt.now().timestamp()
    elif isinstance(_dt_or_ts, int):
        time_asleep = _dt_or_ts - dt.now().timestamp()
    else:
        raise TypeError("given argument is not an integer or datetime object.")

    if time_asleep <= 0:
        return
    
    await sleep(time_asleep)
whole lark
#

You don't need to close the receiver

haughty glen
whole lark
#

You async for the receiver

haughty glen
#

what

whole lark
#

It will wait until you call sender.send_nowait(None)

haughty glen
whole lark
#

In delete_in_background

haughty glen
#

around the whole of it?

whole lark
#

Yeah

haughty glen
# 
async def delete_in_background(pool: Pool, receiver: MemoryObjectReceiveStream[None]):
    async for _ in receiver:
        async with pool.acquire() as conn:
            # Get when the next paste will expire
            req = await conn.execute(
                """
                SELECT id FROM expiries
                ORDER BY expiry_timestamp
                LIMIT 1
                """
            )
            row = await req.fetchone()
        
        if not row:
            # No more records to listen out for
            ... # What goes here now?
        
        await sleep_until(row["delete_at"])

        async with pool.acquire() as conn:
            await conn.execute("DELETE FROM expiries WHERE id = ?", row["id"])
#

what do i do with the receiver now?

whole lark
#

That's it

haughty glen
#

what

#

what do i do after there's no rows?

whole lark
#

Although if you just want to do expiry I'd probably just do it every time you add a new paste

#

Or retrieve a paste, check if it's expired

haughty glen
whole lark
#

Why?

haughty glen
#

it's just sleeping until then

#

and that's what dpy does with its framework, so shrug

whole lark
#

Probably best to just start a background task in the task group for each insert then

haughty glen
whole lark
#

So in this case you want to yield the task group

haughty glen
#

is it cool if i address this in like two hours? i'm going out for a meal shortly

whole lark
#

This channel will close and you'll need to open a new one

haughty glen
whole lark
#

Sounds good then

swift depotBOT
#
Python help channel closed

This help channel has been closed and it's no longer possible to send messages here. If your question wasn't answered, feel free to create a new post in #1035199133436354600. To maximize your chances of getting a response, check out this guide on asking good questions.