Skip to content

Cookies on Jobverse

We use cookies that are needed to sign you in and keep the site working. With your permission we’d also use an analytics cookie to count job views, and Google advertising cookies to show ads. The site works the same whatever you choose. Read the Cookie Policy.

Jobverse
Sign in

Python Backend Developer Top 100 Interview Questions

100 Python backend developer interview questions with structured answers, comparison tables, flow diagrams, worked calculations and runnable code. Covers the language core, concurrency and asyncio, web request handling, databases, caching, testing, security, performance and system design, from the basics to production scenarios.

100 questions

Questions 21-40 of 100

  1. Q21.What does `__slots__` change, and when is it worth using?

    Declaring __slots__ tells Python to store an object's attributes in fixed slots instead of a per-instance dictionary. Instances then use less memory and attribute access is slightly faster.

    • Memory: saving the dictionary matters when you create millions of small objects, such as rows held in memory or events in a queue.
    • Safety: assigning an attribute that is not declared raises an AttributeError, which catches typos.
    • Restrictions: instances cannot gain new attributes dynamically, and every class in an inheritance chain needs its own slots declaration to get the full benefit.

    When it is worth it:

    • Classes with many instances and a fixed set of fields.
    • Performance-sensitive data structures, after profiling shows memory is a real problem.

    For most service code the difference is negligible, so readability wins. Dataclasses can generate slots for you with slots=True, which keeps the code short.

    Example

    The slotted class rejects an undeclared attribute and has no per-instance dictionary.

    class Point:
        __slots__ = ("x", "y")
    
        def __init__(self, x, y):
            self.x = x
            self.y = y
    
    point = Point(1, 2)
    try:
        point.z = 3
    except AttributeError as error:
        print(type(error).__name__)  # AttributeError
    
    print(hasattr(point, "__dict__"))  # False
  2. Q22.How do imports work, and how do you fix a circular import?

    An import first checks the module cache, and only if the module is missing does Python find the file, create the module and run its top-level code. Every later import of the same module returns the cached object.

    flowchart TD
      A["import orders"] --> B{"Already in module cache?"}
      B -->|yes| C["Return cached module"]
      B -->|no| D["Search import path"]
      D --> E["Create empty module and cache it"]
      E --> F["Run module top-level code"]
      F --> G["Return module"]

    Why circular imports fail: module A imports B while A is only half-executed; if B then imports a name from A that A has not defined yet, Python raises an ImportError.

    How to fix one:

    1. Move the shared code into a third module that both can import.
    2. Import the module instead of a name, as in import orders, and use orders.create inside functions, so the lookup happens later.
    3. Move an import inside the function that needs it, as a last resort.
    4. For type hints only, import inside an if TYPE_CHECKING: block.

    A circular import usually signals that two modules have tangled responsibilities, so restructuring is the real cure.

  3. Q23.How do you manage dependencies and reproducible environments for a Python service?

    Isolate each project in its own virtual environment, declare direct dependencies in the project file, and install production from a lock file that pins every package to an exact version.

    1. Virtual environments: the standard venv module gives each project its own interpreter path and packages, so projects never conflict and the system interpreter stays clean.
    2. Direct dependencies: list them in pyproject.toml with compatible version ranges.
    3. Lock file: record the exact resolved version of every direct and indirect package, ideally with hashes, so every build installs identical code.
    4. Separate groups: keep test and development tools out of the production install.
    5. Pin the interpreter: build and test on the same Python version that production uses.
    6. Update deliberately: refresh the lock file on a schedule, run the tests, and review what changed.

    Why it matters:

    • Without a lock file, a new release of an indirect dependency can break a deploy that changed no code.
    • Hashes protect against a tampered package with the expected version number.
  4. Q24.Which built-in operations become slow on large data, and how do you avoid them?

    Several innocent-looking operations take linear time, so inside a loop they turn into quadratic time. Most slow Python services suffer from the wrong data structure rather than from the interpreter itself.

    item in a_lista_list.pop(0)text += piece in a loop
    Scans every element: linear timeShifts every remaining element: linear timeMay copy the growing string on each step
    Use a set or dictionary for membershipUse a deque for queue operationsCollect pieces in a list and join them once

    Other common culprits:

    • Sorting inside a loop when the data could be sorted once.
    • Repeated database queries in a loop instead of one query with a filter.
    • Building large intermediate lists when a generator would stream the data.
    • Deep copies of large structures on every request.

    How to find them: profile first with the standard cProfile module, then look at the functions with the highest cumulative time. Fixing the algorithm usually beats micro-optimizing statements.

    Example

    Checking 10,000 identifiers against a list of 100,000 users does about one billion comparisons in the worst case, because each check scans the list.

    1. With a list: 10,000 × 100,000 = 1,000,000,000 comparisons.
    2. Converting the list to a set once costs about 100,000 operations.
    3. Each set check then takes roughly constant time, so the 10,000 checks cost about 10,000 operations.
    4. The total drops from about a billion steps to about 110,000, which turns seconds into milliseconds.
  5. Q25.How would you find the most common items in a large stream efficiently?

    Count each item as it arrives with a Counter, a dictionary specialized for counting, then ask for the top results. This processes the stream in one pass and keeps memory proportional to the number of distinct items, not the number of events.

    • Counting: counts[item] += 1 works on any iterable, including a generator over a file or a database cursor.
    • Top results: most_common(k) returns the k most frequent items; for a small k on a large set of keys, heapq.nlargest avoids sorting everything.
    • Ties: decide how ties are broken if results must be stable, for example by also sorting by the key.

    When the distinct items do not fit in memory:

    • Count per time window and merge window results.
    • Use an approximate structure such as a count-min sketch, which trades a small, bounded error for fixed memory.
    • Push the aggregation into the database when the data already lives there.

    Example

    Counting request paths in one pass and taking the two most frequent ones.

    import heapq
    from collections import Counter
    
    requests = ["/jobs", "/login", "/jobs", "/apply", "/jobs", "/login", "/health"]
    
    counts = Counter()
    for path in requests:
        counts[path] += 1
    
    print(counts.most_common(2))  # [('/jobs', 3), ('/login', 2)]
    print(heapq.nlargest(2, counts.items(), key=lambda pair: pair[1]))  # [('/jobs', 3), ('/login', 2)]
  6. Q26.What is the global interpreter lock, and how does it affect a backend service?

    The global interpreter lock is a mutex in the default build of the reference interpreter that lets only one thread execute Python bytecode at a time. Threads still run concurrently, but not in parallel on CPU-bound Python code.

    • I/O-bound work is fine: a thread waiting on a socket, file or database releases the lock, so other threads keep running. Most web services spend their time waiting, so threads still help them.
    • CPU-bound work does not scale with threads: parsing, hashing or number crunching in pure Python uses one core, however many threads you start.
    • Native code can release it: many compression, cryptography and numeric extensions release the lock while they work, so they do run in parallel.

    What the lock does not do: it does not make your code thread-safe. An operation such as count += 1 is several bytecode steps, and a thread switch can happen between them.

    How services work around it: run several worker processes, push CPU-heavy work to a process pool or separate workers, or use native libraries. Recent versions also offer an optional free-threaded build without the lock, but it is not the default and some extensions do not support it yet.

  7. Q27.When should you use threads, processes or asyncio in Python?

    Choose by the kind of work: threads and asyncio suit waiting on I/O, while processes suit CPU-heavy work. Many production services combine them, such as several processes each running an event loop.

    ThreadsProcessesAsyncio
    Cannot run Python code in parallel in the default buildRun truly in parallel, each with its own interpreterRuns everything on one thread, switching at each await
    Best for blocking I/O with existing synchronous librariesBest for CPU-heavy work such as image processing or parsingBest for many concurrent network calls with async libraries
    Moderate memory per threadHigh memory and startup cost per processVery low cost per task, so thousands of tasks are cheap
    Risk: race conditions on shared stateRisk: data must be pickled to cross process boundariesRisk: one blocking call stalls every task on the loop

    When to use which:

    • Add threads to synchronous code that waits on several slow calls at once.
    • Use processes when profiling shows the CPU, not waiting, is the bottleneck.
    • Use asyncio for services that hold many slow connections, such as WebSockets or calls to many downstream APIs.
  8. Q28.How does the asyncio event loop run coroutines?

    The event loop is a scheduler running on one thread. It runs a task until that task awaits something that is not ready, then switches to another ready task, and resumes the first one when its I/O completes.

    flowchart TD
      A["Pick next ready task"] --> B["Run until it awaits"]
      B --> C{"Awaited result ready?"}
      C -->|yes| B
      C -->|no| D["Register interest in the socket or timer"]
      D --> E["Wait for operating system readiness events"]
      E --> F["Mark waiting tasks ready"]
      F --> A
    1. Run: the loop runs a coroutine's code until it hits an await on something that must wait.
    2. Suspend: the task's state is saved, and the loop registers the socket or timer it is waiting for.
    3. Poll: when no task is ready, the loop asks the operating system which sockets are readable or writable.
    4. Resume: tasks whose I/O is ready are scheduled again.

    Key consequences:

    • Switching happens only at await, so code between awaits runs without interruption, which removes many race conditions.
    • Any long computation or blocking call between awaits freezes every task on the loop.
  9. Q29.What is the difference between a coroutine, a task and a future in asyncio?

    A coroutine is the work, a task schedules that work on the event loop, and a future is a low-level placeholder for a result that will arrive later. Tasks are in fact a kind of future that drives a coroutine.

    CoroutineTaskFuture
    The object created by calling an async def functionA coroutine scheduled to run on the event loopA placeholder for a result that arrives later
    Does nothing until awaited or scheduledStarts running soon, even if nobody awaits it yetCompleted by callbacks or lower-level code
    Awaiting it runs it in the caller's flow, one step after anotherCan be awaited, cancelled or checked for a resultUsually created by libraries, rarely by application code

    Practical points:

    • Calling a coroutine function without awaiting it does nothing and triggers a "never awaited" warning, a common bug.
    • To run things concurrently, turn coroutines into tasks with asyncio.create_task, a task group or asyncio.gather.
    • Keep a reference to every task you create; an unreferenced task can be garbage-collected, and its errors can go unnoticed.
  10. Q30.How do you run many I/O calls concurrently with asyncio?

    Start the calls as tasks so they wait at the same time, then collect the results. Total time becomes roughly the slowest call instead of the sum of all calls.

    • asyncio.gather: runs awaitables concurrently and returns results in the order you passed them. By default the first exception is raised to the caller, but the other calls keep running.
    • Task groups: asyncio.TaskGroup, available in recent versions, runs tasks in an async with block. If one task fails, the others are cancelled and the errors are raised together as an exception group, so no work leaks.
    • Bounded concurrency: for hundreds of calls, wrap each one with a semaphore so you do not overwhelm the downstream service or run out of connections.

    Good practice:

    • Put a timeout around the whole group, not only individual calls.
    • Prefer task groups for new code, because their failure handling is safer.
    • Use return_exceptions=True with gather only when you really want to handle each failure individually.

    Example

    Three calls of 0.1, 0.2 and 0.3 seconds finish in about 0.3 seconds instead of 0.6, and the task group returns both results.

    import asyncio
    import time
    
    async def fetch(name, delay):
        await asyncio.sleep(delay)
        return f"{name} done"
    
    async def main():
        start = time.perf_counter()
        results = await asyncio.gather(fetch("users", 0.3), fetch("orders", 0.1), fetch("prices", 0.2))
        elapsed = time.perf_counter() - start
        print(results)  # ['users done', 'orders done', 'prices done']
        print(elapsed < 0.5)  # True
    
        async with asyncio.TaskGroup() as group:
            first = group.create_task(fetch("a", 0.1))
            second = group.create_task(fetch("b", 0.1))
        print(first.result(), second.result())  # a done b done
    
    asyncio.run(main())
  11. Q31.What happens if you call blocking code inside an async handler, and how do you fix it?

    A blocking call, such as a synchronous database driver, time.sleep or a heavy computation, holds the event loop's only thread. Until it returns, no other request on that loop makes progress, so one slow call adds its full delay to every concurrent request.

    flowchart TD
      A["Request A calls blocking function"] --> B["Event loop thread is stuck"]
      B --> C["Requests B and C cannot run"]
      C --> D["Latency rises for every client"]
      E["Fix: move blocking call to a thread"] --> F["Loop keeps serving other requests"]

    How to fix it:

    1. Use async libraries for the network and database calls on the hot path.
    2. Offload blocking I/O with await asyncio.to_thread(func, ...), which runs it in a thread pool while the loop continues.
    3. Offload CPU-heavy work to a process pool or separate workers, because a thread would still compete for the interpreter lock.

    How to find it:

    • Enable asyncio debug mode, which logs callbacks that run longer than a threshold.
    • Watch for latency that rises across all endpoints at once while CPU usage stays modest.
  12. Q32.How do you limit concurrency in asyncio?

    Wrap the protected operation in an asyncio.Semaphore, which allows only a fixed number of tasks inside at once. Others wait their turn at async with semaphore, without blocking the event loop.

    Why limit concurrency:

    • Downstream APIs have rate limits and slow down or fail when flooded.
    • Database and HTTP connection pools have a maximum size, and waiting tasks would otherwise time out inside the pool.
    • Thousands of in-flight requests use memory and file descriptors.

    Good practice:

    • Create the semaphore once and share it among all tasks that call the same resource.
    • Size the limit from the downstream capacity or the connection pool size, not by guessing.
    • Combine it with timeouts, so a slow dependency does not hold slots forever.
    • For work arriving continuously, a fixed pool of worker tasks reading from an asyncio.Queue gives the same limit plus backpressure.

    A semaphore limits work inside one process. A limit that must hold across many processes or servers needs a shared mechanism, such as a rate limiter backed by a shared store.

    Example

    Ten calls run, but the peak number inside the protected block never exceeds three.

    import asyncio
    
    async def call_api(item, limit, stats):
        async with limit:
            stats["running"] += 1
            stats["peak"] = max(stats["peak"], stats["running"])
            await asyncio.sleep(0.01)
            stats["running"] -= 1
            return item * 2
    
    async def main():
        limit = asyncio.Semaphore(3)
        stats = {"running": 0, "peak": 0}
        results = await asyncio.gather(*(call_api(i, limit, stats) for i in range(10)))
        print(results)  # [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]
        print(stats["peak"])  # 3
    
    asyncio.run(main())
  13. Q33.How do timeouts and cancellation work in asyncio?

    Cancelling a task raises CancelledError inside it at the await where it is suspended. A timeout is cancellation triggered by a deadline: asyncio.timeout and asyncio.wait_for cancel the work when time runs out and raise TimeoutError to the caller.

    • Cleanup runs: finally blocks and async with exits still execute, so connections and locks are released.
    • Re-raise cancellation: if you catch CancelledError to log or clean up, raise it again; swallowing it breaks timeouts and shutdown.
    • Shielding: asyncio.shield protects an operation that must finish, such as committing a payment record, from the caller's cancellation. Use it sparingly.

    Designing timeouts in a service:

    1. Give every external call a timeout; a call without one can hang forever.
    2. Derive inner timeouts from the request's overall deadline, so the total never exceeds what the client will wait.
    3. Make sure the timed-out work is really stopped, or that it is safe if it completes later.

    Example

    The query is cancelled at its deadline, its cleanup still runs, and the caller sees a timeout.

    import asyncio
    
    async def slow_query(log):
        try:
            await asyncio.sleep(5)
        except asyncio.CancelledError:
            log.append("query cancelled")
            raise
        finally:
            log.append("connection released")
    
    async def main():
        log = []
        try:
            async with asyncio.timeout(0.05):
                await slow_query(log)
        except TimeoutError:
            log.append("timed out")
        print(log)  # ['query cancelled', 'connection released', 'timed out']
    
    asyncio.run(main())
  14. Q34.How do you protect shared state between threads?

    Make every read-modify-write of shared state happen under a lock, or avoid sharing state at all. The interpreter lock does not protect you, because a compound operation such as balance += amount can be interrupted between its steps.

    • threading.Lock: hold it with with lock: around the critical section, and keep that section short.
    • threading.RLock: a lock the same thread can acquire again, useful when locked methods call each other.
    • Thread-safe structures: queue.Queue handles its own locking, so producer and consumer threads can share it safely.
    • Confinement: give each thread its own data, or use threading.local, so nothing needs a lock.
    • Immutability: objects that never change are safe to share without locks.

    Deadlock prevention:

    • Acquire multiple locks in one consistent order everywhere.
    • Never call slow I/O or unknown callbacks while holding a lock.
    • Use timeouts on lock acquisition in code where a hang would be worse than an error.

    Example

    Eight threads each add 10,000 to a counter; with the lock, the final count is always exact.

    import threading
    
    class Counter:
        def __init__(self):
            self.value = 0
            self._lock = threading.Lock()
    
        def increment(self):
            with self._lock:
                self.value += 1
    
    counter = Counter()
    
    def work():
        for _ in range(10_000):
            counter.increment()
    
    threads = [threading.Thread(target=work) for _ in range(8)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()
    
    print(counter.value)  # 80000
  15. Q35.How do you build a producer-consumer pipeline with a thread-safe queue?

    Producers put work items on a queue.Queue, and a fixed pool of consumer threads takes them off and processes them. The queue handles all locking, and a maximum size gives natural backpressure.

    flowchart LR
      A["Producer threads"] -->|"put waits when full"| B["Bounded queue"]
      B -->|get| C["Consumer thread 1"]
      B -->|get| D["Consumer thread 2"]
      C --> E["Results queue"]
      D --> E
    1. Bounded queue: queue.Queue(maxsize=n) makes producers wait when consumers fall behind, so memory cannot grow without limit.
    2. Consumers loop: each worker calls get, processes the item and calls task_done.
    3. Shutdown signal: put one sentinel value per worker, so each one knows when to stop.
    4. Waiting for completion: queue.join blocks until every item put on the queue has been marked done.

    Things to get right:

    • Catch exceptions inside the worker loop, or one bad item kills a thread silently.
    • Collect results through a second queue or a locked list.
    • Use daemon threads only for work that may be abandoned on exit.

    For async code, asyncio.Queue follows the same pattern with awaited calls, and for CPU-heavy work, a process pool replaces the threads.

    Example

    Three workers square six numbers from a bounded queue and stop on a sentinel.

    import queue
    import threading
    
    tasks = queue.Queue(maxsize=5)
    results = []
    results_lock = threading.Lock()
    STOP = object()
    
    def worker():
        while True:
            item = tasks.get()
            if item is STOP:
                tasks.task_done()
                break
            with results_lock:
                results.append(item * item)
            tasks.task_done()
    
    workers = [threading.Thread(target=worker) for _ in range(3)]
    for thread in workers:
        thread.start()
    for number in range(1, 7):
        tasks.put(number)
    for _ in workers:
        tasks.put(STOP)
    
    tasks.join()
    for thread in workers:
        thread.join()
    
    print(sorted(results))  # [1, 4, 9, 16, 25, 36]
  16. Q36.How do you speed up CPU-bound work in a Python service?

    Get the work off the request path and onto multiple cores, after first making the algorithm itself efficient. Threads do not help pure-Python CPU work in the default build, so parallelism needs processes or native code.

    1. Profile first: confirm the time is really CPU, and fix the algorithm or data structure if that is the cause.
    2. Use a process pool: concurrent.futures.ProcessPoolExecutor spreads work across cores. Arguments and results are pickled, so send compact inputs and process work in reasonably large chunks.
    3. Use native implementations: optimized numeric, compression or parsing libraries do the heavy loop in compiled code, often releasing the interpreter lock.
    4. Move it to background workers: for slow jobs, return quickly and process the work from a queue, so web workers stay responsive.
    5. Cache results: if the same inputs repeat, store the output.

    Trade-offs:

    • Each process has its own memory, so many processes multiply memory use.
    • Pickling large objects can cost more than the computation saved.

    Example

    A job takes 400 ms of pure Python CPU time, and 100 jobs arrive at once on a 4-core server.

    1. One thread: 100 × 0.4 = 40 seconds.
    2. Eight threads: still about 40 seconds, because only one thread runs Python bytecode at a time.
    3. A pool of 4 processes: about 40 / 4 = 10 seconds, plus a little time to start processes and pickle data.
    4. Moving these jobs to a background queue also stops them from blocking web requests while they run.
  17. Q37.How do WSGI and ASGI differ?

    Both are standard interfaces between a Python web server and a web application, so any compliant server can run any compliant app. WSGI is the older synchronous interface; ASGI is its asynchronous successor.

    WSGIASGI
    Synchronous: one function call per request, returning the responseAsynchronous: built on coroutines and message events
    Concurrency comes from worker processes and threadsConcurrency comes from the event loop, plus worker processes
    Handles plain request-response HTTPAlso handles WebSockets, long-lived connections and streaming
    Mature and simple for classic web applicationsSuits I/O-heavy services with many slow or open connections

    When to use which:

    • A WSGI app is a good fit when most work is short database queries and libraries are synchronous.
    • An ASGI app fits services that call many downstream APIs, stream data or keep connections open.

    Switching to ASGI only helps if the code on the hot path is genuinely async. An async server running blocking database calls is often slower than a well-sized synchronous one.

  18. Q38.How do worker processes, threads and event loops combine in a production Python server?

    A production deployment usually runs a supervisor process that starts several worker processes, and each worker handles requests with either a thread pool or an event loop. This gets parallelism across cores despite the interpreter lock.

    flowchart LR
      A["Clients"] --> B["Reverse proxy or load balancer"]
      B --> C["Supervisor process"]
      C --> D["Worker process 1"]
      C --> E["Worker process 2"]
      C --> F["Worker process N"]
      D --> G["Thread pool or event loop"]
      E --> G
      F --> G
    • Reverse proxy: terminates TLS, buffers slow clients and serves static files, so Python workers only handle application work.
    • Supervisor: starts workers, restarts any that crash and can recycle workers after a number of requests to contain memory leaks.
    • Workers: each has its own memory and interpreter; process count is usually tied to CPU cores and available memory.
    • Inside a worker: threads suit synchronous apps, and an event loop suits async apps.

    Things to remember:

    • Workers do not share memory, so in-process caches and counters exist separately in each worker.
    • Containers often run one worker process each and scale by adding containers instead.
  19. Q39.How many worker processes does a Python service need for a target load?

    Estimate how many requests are in progress at once, then divide by how many each worker can handle at the same time. Check the result against memory, and confirm it with a load test.

    • concurrent_requests = requests_per_second × average_seconds_per_request
    • workers = ceil(concurrent_requests / requests_per_worker)

    What each term means:

    • requests_per_second: the expected peak arrival rate.
    • average_seconds_per_request: total time a request occupies a worker, including waiting on the database and other services.
    • requests_per_worker: 1 for a single-threaded synchronous worker, the thread count for threaded workers, and much higher for async workers.

    Checks after calculating:

    • Multiply workers by memory per worker and keep the total well below the machine's memory.
    • CPU-bound endpoints are limited by cores, not by this formula.
    • Add headroom of about 30% to 50% for traffic spikes and slow periods, then verify with a load test at the target rate.

    Example

    A synchronous service must handle 200 requests per second, each taking 150 ms on average, mostly waiting on the database.

    1. concurrent_requests = 200 × 0.15 = 30 requests in progress at once.
    2. With 4 threads per worker, workers = ceil(30 / 4) = 8.
    3. Adding 50% headroom gives ceil(8 × 1.5) = 12 workers.
    4. At 250 MB per worker, 12 workers need about 3 GB, so the 8 GB server fits them with room to spare.
  20. Q40.How do you propagate a request ID through async code without passing it everywhere?

    Store it in a contextvars.ContextVar, set once when the request starts. Every function in that request's call chain can read it, and concurrent requests never see each other's value.

    • Isolation: each asyncio task runs in a copy of the context taken when the task was created, so setting the variable in one request does not affect another.
    • Thread support: asyncio.to_thread copies the current context into the worker thread, so logs from offloaded calls keep the ID.
    • Logging integration: a logging filter can read the variable and add it to every log record automatically.

    Typical flow:

    1. Middleware reads the incoming request ID header, or generates a new ID.
    2. It sets the context variable for the rest of the request.
    3. Loggers, HTTP clients and error reports read it, and outgoing calls forward it in a header.

    Thread-local storage does not work for async code, because many requests share one thread. Context variables are the correct tool for both threads and tasks.

    Example

    Two concurrent requests set their own ID, and every log line inside each one shows the right value.

    import asyncio
    import contextvars
    
    request_id = contextvars.ContextVar("request_id", default="-")
    
    def log(message):
        return f"[{request_id.get()}] {message}"
    
    async def load_user(lines):
        await asyncio.sleep(0.01)
        lines.append(log("user loaded"))
    
    async def handle(rid, lines):
        request_id.set(rid)
        lines.append(log("start"))
        await load_user(lines)
    
    async def main():
        lines = []
        await asyncio.gather(handle("req-1", lines), handle("req-2", lines))
        for line in sorted(lines):
            print(line)
    
    asyncio.run(main())
    # Output:
    # [req-1] start
    # [req-1] user loaded
    # [req-2] start
    # [req-2] user loaded