Python Await Async Tutorial with Real Examples and Simple Explanations

Q: What is async/await in Python?

async def defines a coroutine function and await pauses execution until an asynchronous operation completes. This enables concurrent I/O operations without threading, using the asyncio event loop.

Q: How do I run multiple async tasks concurrently?

Use asyncio.gather(task1(), task2()) to run multiple coroutines concurrently. Use asyncio.create_task() to schedule without immediately waiting.

Q: What does 'coroutine was never awaited' mean?

You called an async function without await . Async functions return coroutine objects that must be awaited. Add await before the call or use asyncio.run() from synchronous code.

Q: Can I mix synchronous and asynchronous code?

Yes. Use asyncio.run() to call async from sync. Use loop.run_in_executor() to run blocking functions inside async code without blocking the event loop.

Last Updated: June 01, 2026

Table of Contents

What is async await in Python?
How does async await work in Python?
Cooking Hamburgers Asynchronously and coding the event loop manually
Async Await Code Example in Python
Async Asynchronous Calling Another Async Function Code Example
Async Await Real World Example With Web Crawler in Python
How To Mix Asynchronous And Synchronous Code With Await Async in Python
Conclusion
Related Articles
Frequently Asked Questions

Advanced

The python await and async is one of the more advanced features to help run your programs faster by making sure the CPU is spending as little time as possible waiting and instead as much time as possible working. If ever you see a capable chef, you’ll know what I mean. The chef is not just following a recipe step by step (i.e. working synchronously), the chef is boiling water to cook the pasta , measuring the amount of pasta, chopping tomatoes for the pasta sauce until the water boils etc (i.e. the chef is working asynchronously). The chef is minimizing the time they are waiting idle and always working on a task. That’s the same idea with async and await.

For this tutorial, we will focus on python 3.7 as it has some of the more modern features of await and async. We will call out some of the differences for python 3.4 – 3.6.

Written by Pubs

Python developer and educator with 15+ years building production systems across data engineering, web APIs, and AI tooling. Founder of Python How To Program — 270+ in-depth tutorials covering the modern Python stack.

View all tutorials by Pubs →

What is async await in Python?

The async await keywords help to define in your program which parts need to run sequentially, and which parts may take sometime but other parts of the program can execute while this step completes. A modern example of this is that if you’re downloading a web page it may take a few seconds, while the download is happening you can execute other parts of your program.

How does async await work in Python?

Sometimes the best way to explain something is to show how you would achieve the same thing without the feature.

Continuing with the restaurant theme, suppose you are running a hamburger stall (you’re the waiter and the chef) and it is almost instant to collect payment for a customer and serve the final hamburger, but the most time consuming task is to cooking the beef patty which takes 2 seconds (one could only wish!).

See the below diagram:

Figure 1: Sequentially serving customers at a hamburger stall

In the above diagram:

Step 1: you would first get the order and collect the money from Customer 1
Step 2: you would then put a beef patty on the cook top and then wait for 2 seconds for the beef patty to cook. At the same time, Customer 1 is also waiting for 2 seconds.
Step 3: when the beef patty is cooked, you can then plate this onto a hamburger bun
Step 4: pass the final hamburger to Customer 1
Step 5: You would then start to serve Customer 2 (who has already been waiting 2 seconds for you to serve Customer 1). You can then repeat steps 2-4

With the above approach, Customer 1 would have their burger in about 2 seconds, Customer 2 approx 4 seconds, and then Customer 3 approx 6 seconds.

The equivalent code would be as follows:

import time, datetime, timeit

customer_queue = [ "C1", "C2", "C3" ]

def get_next_customer():
    return customer_queue.pop(0)    #Get the first customer from list

def cook_hamburger(customer):
    start_customer_timer = timeit.default_timer()
    print( f"[{customer}]: Start cooking hamberger for customer")
    time.sleep(2)   # It takes 2 seconds to cook the hamburger
    end_customer_timer = timeit.default_timer()
    print( f"[{customer}]: Finish cooking hamberger for customer.  Total {end_customer_timer-start_customer_timer} seconds\n")

def run_shop():
    while customer_queue:
        curr_customer = get_next_customer()
        cook_hamburger(curr_customer)

def main():
    print('Hamburger Shop')
    start = timeit.default_timer()
    run_shop()
    stop = timeit.default_timer()
    print(f"** Total runtime: {stop-start} seconds ***")

if __name__ == '__main__':
    main()

The code above is fairly straightforward. We have a list of customers that are queuing in the list customer_queue which are being looped under the def run_shop(). For each customer (get_next_customer()), we call cook_hamburger() to cook the hamburger for 2 seconds and wait for it to complete.

Running this code you would get the following output:

As expected, the total runtime for 3 customers is 6 seconds since each customer is served sequentially.

Cooking Hamburgers Asynchronously and coding the event loop manually

Instead of serving the customer and cooking the hamburger for each customer, you can obviously do some of the tasks asynchronously, meaning you can start the task but you don’t have to sit and wait, you can do something else. See the following diagram where the chef/waiter is serving multiple customers and cooking at the same time. It’s not explicitly shown here, but the chef/waiter is constantly checking on the status of the next task and if a task doesn’t require his/her attention they’ll move on to the next task. This process of always looking for something to do is the equivalent of the “event loop”. The Event Loop is a programming construct where the logic is to always look for a task to execute and if there’s a task which will take some time it can release control to the next task in the loop.

Figure 2: *Example of how the event loop works in a real life example – the chef/waiter is always busy!*

In the above example, the following is happening:

Step 1: you would first get the order and collect the money from Customer 1
Step 2: you would then put a beef patty on the cook top and then let it cook, then immediately move on to the next customer while the patty is cooking.
Step 3: you would first get the order and collect the money from Customer 2. You would also check if the first beef patty has completed cooking yet.
Step 4: you would then put another beef patty on the cook top and then let it cook, then immediately move on to the next customer while the patty is cooking.
…
Step 5: When any of the beef patties are done, you would plate it
Step 6: Pass the plated hamburger to the respective customer. Note, in the above example we’ve assumed it to be Customer 1, but it could be any customer depending on which beef patty cooked fully first.
Step 7: When any of the beef patties are done, you would plate it, and server

This is the equivalent of the event loop. The chef/waiter is constantly checking if it needs to serve the customer or check on the hamburgers which are cooking. When there’s a hamburger is placed on the stove and we need to wait 2 seconds, the chef/waiter moves to the next task and does not wait for the 2 seconds to complete. When the hamburger is done, it is then served to the customer.

How can this be done programatically? Glad you asked:

import time ,datetime, timeit

customer_queue = [ "C1", "C2", "C3" ]
hamburger_queue = []

def get_next_customer():
    if customer_queue: return customer_queue.pop(0)    #Get the first customer from list
    return None 

def start_cooking_hamburger(customer):
    print( f"[{customer}]: Start cooking hamberger for customer")
    hamburger = { "customer":customer, "start_cooking_time": timeit.default_timer(), "cooked":False}
    hamburger_queue.append( hamburger )

def check_hamburger_status():
    curr_timer = timeit.default_timer()

    #Check if it's cooking, but release control
    for index, hamburger in enumerate(hamburger_queue):         
        elapsed_time = curr_timer-hamburger['start_cooking_time']
        if elapsed_time > 2: #2 second has passed for hamrburger to cook
            print( f"[{hamburger['customer']}]: Finish cooking hamberger for customer.  Total {elapsed_time} seconds\n")
            del hamburger_queue[ index].  #delete from list to mark as done

def run_shop():
    while customer_queue or hamburger_queue:        #Event loop
        curr_customer = get_next_customer()
        if curr_customer: start_cooking_hamburger(curr_customer)
        check_hamburger_status()

def main():
    print('Hamburger Shop')
    start = timeit.default_timer()
    run_shop()
    stop = timeit.default_timer()
    print(f"** Total runtime: {stop-start} seconds ***")

if __name__ == '__main__':
    main()

The output of the code is as follows:

*Output running asynchronously – notice the runtime of 2 seconds compared to the 6 seconds in the synchronsous method*.

So there’s a few things happening here:

There’s a new list called hamburger_queue[] which is keeping track of each hamburger that is being cooked
The event loop is the while customer_queue or hamburger_queue within the run_shop() function
We have a new function called start_cooking_hamburger() which helps to keep track of the task to cooking starting. Why is this needed? Well in the past we would simply wait for a given task. Now, since we are doing something else while we wait, we need to remember a few things to come back to the task
We also have a new function called check_hamburger_status() which checks the status of each hamburger being cooked (i.e. item in hamburger_queue[]), and if it is cooked (i.e. 2 seconds have passed), then it is considered complete

You may notice in the output that Customer 3 was in fact served before Customer 2. This is because that the execution order is not guarantee.

How To Get CPU Core Usage with psutil in Python

by Pubs | Aug 16, 2026 | Automation

Intermediate

Your server is running slow, but top shows average CPU at 45% — nothing alarming. Then a colleague points out that core 3 has been pinned at 100% for the last hour while the other seven cores sit idle. A single-threaded bottleneck is strangling your app, invisible to anyone watching only the aggregate number. This is exactly the kind of problem you cannot catch without per-core monitoring, and Python makes it surprisingly easy to build.

The psutil library gives you cross-platform access to CPU usage per core, per-core clock frequency, per-core time breakdowns (user, system, idle), and memory statistics — all in a few lines of Python. It works identically on Windows, macOS, and Linux without requiring root access or system-specific tools like top, htop, or Task Manager. Install it once with pip and you are ready to go.

In this article we will cover everything you need to build a CPU monitoring tool with psutil. We start with a Quick Example so you get per-core numbers immediately. Then we dig into cpu_percent(), physical vs logical core counts, per-core frequency with cpu_freq(), time breakdowns with cpu_times(), memory monitoring, and threshold-based alerting. By the end you will have a real-time terminal dashboard you can point at any machine.

Getting Per-Core CPU Usage: Quick Example

Let us start with the most useful function in psutil for this task. The key is the percpu=True flag on cpu_percent() — without it you get one aggregate number; with it you get a list of percentages, one per logical core.

# quick_cpu_check.py
import psutil
import time

# Pass interval=1 to measure over a 1-second window (recommended)
# percpu=True returns a list -- one value per logical CPU core
core_usage = psutil.cpu_percent(interval=1, percpu=True)

print(f"Logical cores detected: {len(core_usage)}")
print()

for i, pct in enumerate(core_usage):
    bar = "#" * int(pct / 5)
    print(f"  Core {i:>2}: {pct:5.1f}%  [{bar:<20}]")

print()
print(f"  Overall: {psutil.cpu_percent(interval=None):.1f}%")

Output:

Logical cores detected: 8

  Core  0:  23.4%  [####                ]
  Core  1:   8.1%  [#                   ]
  Core  2:  91.3%  [##################  ]
  Core  3:   6.2%  [#                   ]
  Core  4:  12.7%  [##                  ]
  Core  5:   9.4%  [#                   ]
  Core  6:  17.6%  [###                 ]
  Core  7:   5.0%  [#                   ]

  Overall: 21.7%

The output instantly reveals that core 2 is at 91% while the overall average looks benign at 21.7%. That discrepancy is exactly what aggregate monitoring misses. The interval=1 parameter tells psutil to collect a sample, wait one second, collect another, and return the difference -- this gives you a meaningful measurement rather than a snapshot that could be zero. The len(core_usage) check tells you how many logical cores the machine has, which varies from 2 on a budget laptop to 128 on a high-end server.

The rest of this article explains how each piece works, adds frequency and memory data, and builds toward a live refreshing terminal dashboard. Read on for the details, or jump straight to the Real-Life Example if you want the full script now.

What is psutil and Why Use It?

psutil (process and system utilities) is a cross-platform library for retrieving information on running processes and system utilization -- CPU, memory, disks, network, and sensors. It wraps the underlying OS interfaces (/proc on Linux, sysctl on macOS, Win32 API on Windows) so your Python code runs unchanged on all three platforms.

The alternative to psutil is platform-specific shell commands: mpstat -P ALL 1 on Linux, sysctl hw.perflevel0.physicalcpu on macOS, or WMI queries on Windows. You could parse their output with subprocess, but you would need separate code paths for each OS and your script would break every time the command output format changes. psutil solves all of that.

Method	Platform	Root Required	Per-Core Data	Python API
psutil	Windows / macOS / Linux	No	Yes	Yes -- clean objects
mpstat	Linux only	No	Yes	Parse subprocess output
top / htop	Unix-like	No	Yes	No -- interactive only
WMI	Windows only	Admin for some	Partial	Via pywin32
/proc/stat	Linux only	No	Yes	Manual file parsing

Install psutil with pip -- it has no dependencies and compiles quickly:

# install_psutil.sh
pip install psutil

Once installed you can import it and immediately start querying system metrics. The sections below walk through each function you need for CPU monitoring.

Python developer at terminal showing per-core CPU usage bars with psutil — cpu_percent(percpu=True) -- because one number hides too much.

Logical vs Physical Cores: What cpu_count() Returns

Before diving deeper into usage numbers, it helps to understand what "core" actually means here. Modern CPUs expose more logical cores than they have physical cores because of hyperthreading (Intel) or SMT (AMD). A 4-core chip with hyperthreading shows up as 8 logical cores. psutil lets you query both counts.

# core_count.py
import psutil

logical = psutil.cpu_count(logical=True)   # includes hyperthreads
physical = psutil.cpu_count(logical=False)  # physical cores only

print(f"Physical cores:  {physical}")
print(f"Logical cores:   {logical}")
print(f"Hyperthreading:  {'Yes' if logical > physical else 'No'}")
print(f"HT ratio:        {logical // physical}x" if physical else "")

Output:

Physical cores:  4
Logical cores:   8
Hyperthreading:  Yes
HT ratio:        2x

The number of items in the list returned by cpu_percent(percpu=True) always matches cpu_count(logical=True) -- you get one entry per logical core. Physical core count matters for workloads that benefit from true parallelism (CPU-bound Python processes, for example) vs workloads that are mostly I/O-bound and can share a core fine. Knowing the physical count also helps you interpret the per-core usage: if logical cores 0 and 1 are both busy, that is likely one physical core under full load.

Per-Core Frequency with cpu_freq()

CPU frequency tells you whether a core is running at full speed or has been throttled by thermal limits. Modern processors use dynamic frequency scaling: they boost above the rated speed when the workload demands it (and the chip is cool enough), and throttle down to save power or prevent overheating.

# cpu_frequency.py
import psutil

# percpu=True returns a list of scpufreq namedtuples
freqs = psutil.cpu_freq(percpu=True)

if freqs:
    print(f"{'Core':<8} {'Current MHz':>12} {'Min MHz':>10} {'Max MHz':>10}")
    print("-" * 44)
    for i, f in enumerate(freqs):
        print(f"Core {i:<3}  {f.current:>10.0f}   {f.min:>9.0f}  {f.max:>9.0f}")
else:
    # Some Linux VMs do not expose per-core frequency
    overall = psutil.cpu_freq()
    print(f"Per-core freq not available. Overall: {overall.current:.0f} MHz")

Output:

Core     Current MHz    Min MHz    Max MHz
--------------------------------------------
Core 0      3600        800       4200
Core 1      4100        800       4200
Core 2      4200        800       4200
Core 3      3200        800       4200
Core 4      3800        800       4200
Core 5      4000        800       4200
Core 6      4200        800       4200
Core 7      2900        800       4200

A core sitting at its maximum frequency (4200 MHz here) that also shows high CPU usage is healthy -- it is working hard and boosting as designed. A core showing high CPU usage but stuck at minimum frequency (800 MHz) is likely being throttled due to heat, and you have a cooling problem rather than a workload problem. The defensive check for if freqs: is important: some virtualized Linux environments do not expose per-core frequency and return an empty list.

Developer diagnosing a throttled CPU core with psutil cpu_freq — Throttled cores are the silent killers. psutil.cpu_freq() exposes them.

Per-Core Time Breakdown with cpu_times()

CPU usage percentage tells you HOW MUCH a core is working, but not what it is doing. cpu_times() breaks the time a CPU has spent into categories: user space (your code), kernel space (system calls), idle, and on Linux you also get I/O wait and steal time (from hypervisor overhead in VMs).

# cpu_times_breakdown.py
import psutil

times = psutil.cpu_times(percpu=True)

print(f"{'Core':<6} {'User%':>7} {'Sys%':>7} {'Idle%':>7} {'IOWait%':>9}")
print("-" * 40)

for i, t in enumerate(times):
    total = t.user + t.system + t.idle + getattr(t, 'iowait', 0.0)
    if total == 0:
        continue
    user_pct   = t.user   / total * 100
    sys_pct    = t.system / total * 100
    idle_pct   = t.idle   / total * 100
    iowait_pct = getattr(t, 'iowait', 0.0) / total * 100
    print(f"Core {i:<1}  {user_pct:>7.1f} {sys_pct:>7.1f} {idle_pct:>7.1f} {iowait_pct:>9.1f}")

Output:

Core   User%    Sys%   Idle%   IOWait%
----------------------------------------
Core 0   18.2     4.1    77.7       0.0
Core 1    6.5     1.6    91.9       0.0
Core 2   88.4     2.9     8.7       0.0
Core 3    5.1     1.1    93.8       0.0
Core 4   11.3     0.8    87.9       0.1
Core 5    8.7     0.9    90.4       0.0
Core 6   15.2     1.4    83.4       0.0
Core 7    4.2     0.6    95.2       0.0

Note the getattr(t, 'iowait', 0.0) pattern. The iowait field only exists on Linux; using getattr with a default keeps the code portable to macOS and Windows. A core with high user% is running application code. High sys% means lots of system calls (file I/O, socket operations). High iowait% means the core is waiting on storage -- often a sign that your database or file access is the real bottleneck, not CPU.

Memory Monitoring: virtual_memory()

CPU monitoring is rarely useful in isolation -- memory pressure often causes CPU spikes as the OS spends cycles on swapping. Adding memory data to your monitor gives a more complete picture.

# memory_check.py
import psutil

mem = psutil.virtual_memory()
swap = psutil.swap_memory()

def fmt_bytes(n):
    for unit in ('B', 'KB', 'MB', 'GB', 'TB'):
        if n < 1024:
            return f"{n:.1f} {unit}"
        n /= 1024
    return f"{n:.1f} PB"

print("RAM:")
print(f"  Total:     {fmt_bytes(mem.total)}")
print(f"  Available: {fmt_bytes(mem.available)}")
print(f"  Used:      {fmt_bytes(mem.used)}  ({mem.percent:.1f}%)")
print(f"  Buffers:   {fmt_bytes(getattr(mem, 'buffers', 0))}")
print(f"  Cached:    {fmt_bytes(getattr(mem, 'cached', 0))}")
print()
print("Swap:")
print(f"  Total:     {fmt_bytes(swap.total)}")
print(f"  Used:      {fmt_bytes(swap.used)}  ({swap.percent:.1f}%)")

Output:

RAM:
  Total:     15.9 GB
  Available: 9.3 GB
  Used:      6.1 GB  (38.7%)
  Buffers:   312.0 MB
  Cached:    4.2 GB

Swap:
  Total:     2.0 GB
  Used:      0.0 MB  (0.0%)

The mem.available field is the most actionable metric here -- it is not the same as mem.total - mem.used. Available includes memory that is currently used for caches but can be reclaimed immediately by applications. If mem.available drops near zero while swap.percent climbs, your machine is under genuine memory pressure and performance will degrade. The getattr calls on buffers and cached guard against Windows, which does not expose those fields.

Developer overloading server RAM visualizing memory pressure — swap.percent > 0 is the system's way of asking for help.

Threshold Alerting: Raising Warnings When Cores Spike

Collecting metrics is only useful if something reacts to them. The next step is adding threshold checks so your monitoring code can trigger an alert, write to a log file, or send a notification when a core crosses a usage limit you define.

# cpu_alerts.py
import psutil
import time
import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%H:%M:%S",
)

CPU_WARN_PCT  = 70.0   # warn if any single core exceeds this
CPU_CRIT_PCT  = 90.0   # critical if any core exceeds this
MEM_WARN_PCT  = 80.0   # warn if RAM usage exceeds this
CHECK_INTERVAL = 5     # seconds between checks

def check_once():
    per_core = psutil.cpu_percent(interval=1, percpu=True)
    mem = psutil.virtual_memory()

    for i, pct in enumerate(per_core):
        if pct >= CPU_CRIT_PCT:
            logging.critical("Core %d at %.1f%% -- CRITICAL", i, pct)
        elif pct >= CPU_WARN_PCT:
            logging.warning("Core %d at %.1f%% -- high usage", i, pct)

    if mem.percent >= MEM_WARN_PCT:
        logging.warning("RAM at %.1f%% -- available: %.1f GB",
                        mem.percent, mem.available / 1e9)

if __name__ == "__main__":
    logging.info("Starting CPU/memory monitor (Ctrl+C to stop)")
    try:
        while True:
            check_once()
            time.sleep(CHECK_INTERVAL)
    except KeyboardInterrupt:
        logging.info("Monitor stopped.")

Output:

09:14:01 [INFO] Starting CPU/memory monitor (Ctrl+C to stop)
09:14:02 [WARNING] Core 2 at 73.5% -- high usage
09:14:07 [CRITICAL] Core 2 at 94.1% -- CRITICAL
09:14:12 [CRITICAL] Core 2 at 98.7% -- CRITICAL
09:14:17 [INFO] Monitor stopped.

Using the standard logging module rather than print() means you can redirect this output to a file with one line change (filename="monitor.log" in the basicConfig call), or hook it into any structured logging pipeline. The CHECK_INTERVAL constant separated from cpu_percent(interval=1) is intentional -- the interval on cpu_percent controls measurement accuracy, while CHECK_INTERVAL controls how often you act on the results.

Real-Life Example: Live Terminal CPU Dashboard

Let us combine everything into a dashboard that refreshes in place every two seconds, showing per-core bars, frequency, and memory -- all in one compact terminal view.

Developer satisfied at terminal showing live CPU usage dashboard built with psutil — A dashboard that actually tells you which core is screaming.

# cpu_dashboard.py
import psutil
import time
import os

CPU_WARN  = 70.0
CPU_CRIT  = 90.0
REFRESH   = 2.0   # seconds between refreshes

def color(pct):
    """Return ANSI color code based on usage percentage."""
    if pct >= CPU_CRIT:
        return "\033[91m"   # bright red
    if pct >= CPU_WARN:
        return "\033[93m"   # yellow
    return "\033[92m"       # green

RESET = "\033[0m"

def make_bar(pct, width=24):
    filled = int(pct / 100 * width)
    return "#" * filled + "-" * (width - filled)

def render():
    os.system("cls" if os.name == "nt" else "clear")
    print("=" * 56)
    print("  psutil CPU Dashboard -- press Ctrl+C to exit")
    print("=" * 56)

    per_core = psutil.cpu_percent(interval=1, percpu=True)
    freqs    = psutil.cpu_freq(percpu=True) or []
    mem      = psutil.virtual_memory()
    logical  = psutil.cpu_count(logical=True)
    physical = psutil.cpu_count(logical=False)

    print(f"  Cores: {physical} physical / {logical} logical\n")

    for i, pct in enumerate(per_core):
        freq_str = ""
        if i < len(freqs):
            freq_str = f"  {freqs[i].current:>5.0f} MHz"
        bar = make_bar(pct)
        c = color(pct)
        print(f"  Core {i:>2}: {c}[{bar}]{RESET} {pct:5.1f}%{freq_str}")

    avg = sum(per_core) / len(per_core) if per_core else 0
    print(f"\n  Avg:    [{make_bar(avg)}] {avg:5.1f}%")
    print()

    mem_bar = make_bar(mem.percent, width=24)
    mc = color(mem.percent)
    avail_gb = mem.available / 1e9
    print(f"  RAM:    {mc}[{mem_bar}]{RESET} {mem.percent:5.1f}%  "
          f"({avail_gb:.1f} GB free)")

    swap = psutil.swap_memory()
    if swap.total > 0:
        swap_bar = make_bar(swap.percent, width=24)
        sc = color(swap.percent)
        print(f"  Swap:   {sc}[{swap_bar}]{RESET} {swap.percent:5.1f}%")

    print()
    print(f"  Updated every {REFRESH}s -- {time.strftime('%H:%M:%S')}")
    print("=" * 56)

if __name__ == "__main__":
    try:
        while True:
            render()
            time.sleep(REFRESH)
    except KeyboardInterrupt:
        print("\nDashboard stopped.")

Output (sample frame):

========================================================
  psutil CPU Dashboard -- press Ctrl+C to exit
========================================================
  Cores: 4 physical / 8 logical

  Core  0: [######------------------]  25.4%   3600 MHz
  Core  1: [#-----------------------]   8.1%   2900 MHz
  Core  2: [######################--]  91.3%   4200 MHz
  Core  3: [#-----------------------]   6.2%   3100 MHz
  Core  4: [###---------------------]  12.7%   3400 MHz
  Core  5: [##----------------------]   9.4%   3200 MHz
  Core  6: [###---------------------]  17.6%   3800 MHz
  Core  7: [#-----------------------]   5.0%   2800 MHz

  Avg:    [####--------------------]  22.0%

  RAM:    [############------------]  51.2%  (7.8 GB free)
  Swap:   [------------------------]   0.0%

  Updated every 2s -- 09:17:44
========================================================

The os.system("cls" if os.name == "nt" else "clear") call clears the terminal before each refresh, giving the appearance of an in-place update rather than scrolling output. The ANSI color codes turn critical cores red and high-usage cores yellow in any terminal that supports them (macOS Terminal, Linux terminals, Windows Terminal). To log to a file instead of the terminal, replace the render() call with the check_once() pattern from the alerting section. You can also extend this script by adding disk I/O stats with psutil.disk_io_counters(perdisk=True) or network throughput with psutil.net_io_counters(pernic=True).

Frequently Asked Questions

Why does cpu_percent() return 0.0 when I call it with no arguments?

The first call to psutil.cpu_percent() with no interval and no previous call in the same process always returns 0.0. psutil calculates CPU usage as the difference between two samples taken some time apart. The first call just sets the baseline; the second call (or a call with interval=N) returns the actual measurement. Always use interval=1 (or at least 0.1) for accurate readings, or call the function once at startup to prime it and then call it again after a small sleep.

When should I use logical=True vs logical=False in cpu_count()?

Use cpu_count(logical=True) when you want to know how many workers to create for I/O-bound tasks -- more logical cores means more threads can be useful. Use cpu_count(logical=False) for CPU-bound work where you spawn Python processes -- extra logical cores from hyperthreading rarely help CPU-bound code and can actually hurt throughput by competing for the same physical core resources. When in doubt, benchmark both: run your workload with physical workers and with logical workers and compare wall-clock time.

Does psutil need root/admin privileges?

No -- reading CPU usage percentages, frequencies, core counts, and memory stats does not require elevated permissions on Windows, macOS, or Linux. Some psutil functions DO require root, such as reading per-process memory maps or certain sensor temperatures (psutil.sensors_temperatures()). For a pure CPU and memory monitoring script like the one in this article, you can run as a regular user. If you get a psutil.AccessDenied exception, check which specific function triggered it -- it is almost certainly a process-level function, not a system-level one.

cpu_freq(percpu=True) returns an empty list on my Linux VM. What is wrong?

This is expected behavior on many virtualized Linux environments. The guest OS does not always have access to the host CPU's frequency scaling information. The psutil.cpu_freq() function reads from /sys/devices/system/cpu/cpu*/cpufreq/ on Linux, which may not be populated by the hypervisor. Some cloud VMs (AWS, GCP, Azure) intentionally withhold this data. The safe approach is to always check if freqs: before iterating, and fall back to a single aggregate call (psutil.cpu_freq(percpu=False)) or simply skip the frequency column. The CPU usage percentage from cpu_percent() remains accurate even when frequency data is unavailable.

Does this code work on Windows without any changes?

Yes, with one small caveat: the ANSI color codes in the dashboard script require Windows 10 version 1607 or later with Windows Terminal or a VT100-compatible terminal. The standard Windows Command Prompt (cmd.exe) on older Windows versions does not render ANSI codes and will display them as literal characters like [91m. You can guard against this by wrapping the ANSI output in a try/except or by using the colorama library (pip install colorama), which translates ANSI codes to Win32 console calls. Everything else -- cpu_percent(), cpu_count(), cpu_freq(), virtual_memory(), and swap_memory() -- works identically on Windows.

Can I get CPU temperature with psutil?

On Linux and some macOS hardware, yes: psutil.sensors_temperatures() returns a dictionary of sensor readings grouped by device name. The key for CPU cores is usually 'coretemp' or 'k10temp' depending on the chip. Each entry has current, high, and critical temperature values in Celsius. This function is not available on Windows -- psutil simply does not expose it there because the Windows thermal sensor APIs require platform-specific third-party libraries. On unsupported platforms the call raises AttributeError, so always check hasattr(psutil, 'sensors_temperatures') before using it.

Conclusion

psutil makes per-core CPU monitoring a matter of two function calls. cpu_percent(interval=1, percpu=True) gives you a list of usage values -- one per logical core -- that reveals the imbalances a single aggregate number would hide. cpu_count(logical=True/False) tells you whether extra cores come from hyperthreading or are genuine physical cores. cpu_freq(percpu=True) shows whether cores are boosting or being throttled. cpu_times(percpu=True) breaks usage down into user, system, and iowait time so you know whether CPU cycles are spent on application code, kernel calls, or waiting on storage. And virtual_memory() and swap_memory() round out the picture by capturing memory pressure alongside CPU load.

Extend the dashboard by adding psutil.disk_io_counters(perdisk=True) for storage throughput, psutil.net_io_counters(pernic=True) for network stats, or hook the alert thresholds into a notification service like Slack or PagerDuty. You could also export metrics to a time-series database like Prometheus by wrapping the psutil calls in a Flask endpoint and adding a Prometheus client. The psutil documentation at psutil.readthedocs.io covers every available function in depth.

For deeper exploration, the Python Scalene profiler article shows how to go beyond monitoring into detailed line-level CPU and memory profiling within your own code, and the Python task automation guide covers scheduling monitoring scripts to run on a cron job.

« Older Entries

Async Await Code Example in Python

In the previous section we created an asynchronous version manually. Here’s the same outcome but written with the async await syntax. As you’ll notice it is very similar to the original synchronous version:

import time, datetime, time
import asyncio

import time, datetime, timeit

customer_queue = [ "C1", "C2", "C3"  ]

def get_next_customer():
    return customer_queue.pop(0)    #Get the first customer from list

async def cook_hamburger(customer):     
    start_customer_timer = timeit.default_timer()
    print( f"[{customer}]: Start cooking hamberger for customer")
    await asyncio.sleep(2)   # Sleep but release control
    end_customer_timer = timeit.default_timer()
    print( f"[{customer}]: Finish cooking hamberger for customer.  Total {end_customer_timer-start_customer_timer} seconds\n")

async def run_shop():
    cooking_queue = []

    while customer_queue:
        curr_customer = get_next_customer()
        cooking_queue.append(  cook_hamburger(curr_customer)  )   #this returns a task only

    #cooking_queue[] has all the async tasks
    await asyncio.gather( *cooking_queue )      #Run all in parallel

def main():
    print('Hamburger Shop')
    start = timeit.default_timer()

    asyncio.run( run_shop() )           #Start the event loop

    stop = timeit.default_timer()
    print(f"** Total runtime: {stop-start} seconds ***")

if __name__ == '__main__':
    main()

Output as follows:

Let’s walk through the code:

Firstly, the async await is available from the library asyncio hence the import asyncio
There’s funny set of async keywords which precede the def run_shop() and the def cook_hamburger(customer) functions. In addition the run_shop() is no longer called directly, instead it is called with a asyncio.run( run_shop() ) function call. So here’s what is happening:
- The asyncio.run() function is the trigger for the so-called event loop. It continues to run forever until all the tasks given to it are completed. You must pass it a function with the async def... prefix hence why run_shop() has the async prefix
- In the async def run_shop() function call, the code iterates while there are customers in the queue to process, and then there’s a call to cook_hamburger(curr_customer) for each customer. A direct call to the customer does not actually call the function but instead creates a task to execute this. That is what the async tells the compiler – that when called directly, return a task.
- At the end of the function code in def run_shop() there’s a call to function await asyncio.gather( *cooking_queue). There’s a few things going on here:
  - The await keywords indicates that you need wait for the work to complete but python can do something else in the meantime
  - The call to gather() actually executes all the tasks given to it as a parameter collectively as a group and then returns the results sequentially (please note that the order of the tasks being executed may be random)
  - The *customer_queue simply expands the list into a list of parameter items. So for example if customer_queue[] == [ '1', '2', '3'] then the gather( *customer_queue) would be the same as gather( '1', '2', '3').
- When the await asyncio.gather( *customer_queue ) is called, the await keyword releases control to any activities that are pending and one of them would be to the calls to function cook_hamburger() which was added to the customer_queue list. Hence calls to cook_hamburger() would be triggered.
- Within cook_hamburger() there is also an await asyncio.sleep(2). This simply waits for 2 seconds, however, it does not force the program to wait for the 2 seconds to complete, instead the await keyword releases python to do something else in the meantime. This is similar to step 3 in Figure 2 where the chef/waiter puts the hamburger on the grill, but then doesn’t wait for the 2 second but instead does something else (i.e. serve the next customer)
The asyncio.run() are new keywords as part of python 3.7. In older versions of python you may see the following but it is the same as simply running asyncio.run( run_shop() ) :
- loop = asyncio.get_event_loop()
- loop.run_until_complete(run_shop())
- loop.close()
As you will notice, this is very similar to the synchronous code that covers Figure 1 above. This is the beauty of async/await

So remember, whenever there’s an await then that means python pauses at that point for that task to complete but then also releases python to do something else. That’s how the performance improvement occurs. In this example, the runtime of this is 2 seconds instead of the sequential 6 seconds!

Async Asynchronous Calling Another Async Function Code Example

Suppose you want t also call another async function once your first async function is completed – how do you go about this? Remember the rule, if you want to run something asynchronously, you have to use the await keyword, and that the function you’re calling has to be defined with async def ...

To continue with the restaurant theme, suppose that after the hamburger is cooked you ask an assistant to put the hamburger into a takeaway bag which takes 1 second. This is also another task that you need not ‘block’ and wait for it to complete. Hence, this action can be put into a function which is defined as an async. Here’s what the code can look like:

import time, datetime, time
import asyncio

customer_queue = [ "C1", "C2", "C3" ]

def get_next_customer():
    return customer_queue.pop(0)    #Get the first customer from list

async def cook_hamburger(customer):     
    start_customer_timer = timeit.default_timer()
    print( f"[{customer}]: Start cooking hamberger for customer")
    await asyncio.sleep(2)   # Sleep but release control
    end_customer_timer = timeit.default_timer()
    print( f"[{customer}]: Finish cooking hamberger for customer.  Total {end_customer_timer-start_customer_timer} seconds")
    await put_hamburger_in_takeaway_bag( customer )

async def put_hamburger_in_takeaway_bag( customer):
    start_customer_timer = timeit.default_timer()
    print( f"[{customer}]: Start packing hamberger")
    await asyncio.sleep(1)   # It takes 2 seconds to cook the hamburger
    end_customer_timer = timeit.default_timer()
    print( f"[{customer}]: Finish packing hamberger.  Total {end_customer_timer-start_customer_timer} seconds\n")

async def run_shop():
    cooking_queue = []

    while customer_queue:
        curr_customer = get_next_customer()
        cooking_queue.append( cook_hamburger(curr_customer) )   #Get each of the event loops
    await asyncio.gather( *cooking_queue )      #Run all in parallel

def main():
    print('Hamburger Shop')
    start = timeit.default_timer()
    asyncio.run( run_shop() )           #Start the event loop 
    stop = timeit.default_timer()
    print(f"** Total runtime: {stop-start} seconds ***")

if __name__ == '__main__':
    main()

The output would be:

See how once the hamburger is cooked (e.g. [C1]: Finish cooking hamburger for customer. Total 2.000924572115764 seconds), then immediately afterwards you have the [C1]: Start packing hamburger step but also gets called asynchronously.

Async Await Real World Example With Web Crawler in Python

One difficulty in learning Async / Await is that many examples provided simply provide the asyncio.sleep() as an example which is helpful to understand the concept, but not very helpful when you want to make something more useful. Let’s try a more complex example where you want to get some stock data from finance.yahoo.com and then, for that same stock, you also get the first 3 newspaper articles from news.google.com in the last 24 hours.

Now one thing you will realise is that await only works with functions that are defined as async. So you cannot call any function with await. Why? Well recall that when you call await you are expecting a function to return a task and not actually call the function, hence that function needs to be defined as async in order to tell python that it returns a task to be executed at the next available time.

Let’s see the synchronous version of the code:

import asyncio, requests, timeit
from bs4 import BeautifulSoup
from pygooglenews import GoogleNews

stock_list = [ "TSLA", "AAPL"]

def get_stock_price_data(stock):
    print(f"-- getting stock data for {stock}")
    data = {"stock":stock, "price_open":0, "price_close":0 }
    stock_page = requests.get( 'https://finance.yahoo.com/quote/' + stock, headers={'Cache-Control': 'no-cache',  "Pragma": "no-cache"})

    soup = BeautifulSoup(stock_page.text, 'html.parser')
    #<fin-streamer active="" class="Fw(b) Fz(36px) Mb(-4px) D(ib)" data-field="regularMarketPrice" data-pricehint="2" data-symbol="TSLA" data-test="qsp-price" data-trend="none" value="759.63">759.63</fin-streamer>
    data['price_close'] = soup.find('fin-streamer', attrs={"data-symbol":stock, "data-field":"regularMarketPrice"} ).text

    #<td class="Ta(end) Fw(600) Lh(14px)" data-test="OPEN-value">723.25</td>
    data['price_open'] = soup.find( attrs={"data-test":"OPEN-value"}).text

    return data

def get_recent_news(stock):
    print(f"-- getting news data for {stock}")
    gn = GoogleNews()
    search = gn.search(f"stocks {stock}", when = '24h')
    news = search['entries'][0:3]
    return news

def print_stock_update(stock, data, news):
    print(f"Stock:{ stock }")
    price_change = 0
    if int(float(data['price_open'])) != 0: price_change = round( 100 * ( float( data['price_close'])/float(data['price_open'])-1), 2)
    print(f"Open Price:{data['price_open']} Close Price:{data['price_close']} Change:{price_change}% ")
    print("Latest News:")
    for news_item in news:        
        print( f"{news_item.published}:{news_item.source.title} - {news_item.title}" )
    print("\n")

def process_stocks():
    for stock in stock_list:
        data = get_stock_price_data( stock )
        news=[]
        news = get_recent_news( stock )
        print_stock_update(stock, data, news)

if __name__ == '__main__':
    start_timer = timeit.default_timer()
    process_stocks()
    end_timer = timeit.default_timer()

    print(f"** Total runtime: {end_timer-start_timer} seconds ***")

Output as follows:

So what’s happening here. Well, you are looping through two stocks TSLA and AAPL, and for each stock the following happens sequentially:

A call to data = get_stock_price_data( stock ) occurs in order to make a call to requests.get( 'https://finance.yahoo.com/quote/' + stock) to get the HTML page for the TSLA stock. Effectively, this page: https://finance.yahoo.com/quote/TSLA
Next we use BeautifulSoup() in order to find the HTML snippet that contains the stock price data for the opening price and the closing price:

After the call to yahoo is complete, then there’s a call to news = get_recent_news( stock ) which uses the module pygooglenews to get the latest google news. In fact we have used this function in our previous Twitter Bot article.
Once this is all done, that output is printed out with the call to print_stock_update(stock, data, news)

Clearly this could be called asynchronously as we are looping each time for each stock, and then also the call to get the stock data is independent to getting the news data. However, one thing has to happen sequentially is the print_stock_update(stock, data, news) which has to wait for both the async calls to complete.

One wait to try is to simply call the website download with:

stock_page = await requests.get( 'https://finance.yahoo.com/quote/' + stock, headers={'Cache-Control': 'no-cache',  "Pragma": "no-cache"})

However, you will get the following error:

The reason is, as you may have guessed, is that the requests.get() is not created with the async def... construct and hence cannot be called asynchronously.

What you can do however is to use another ‘get’ web page module called httpx. This function is defined with async def... and can be called similar to requests. That same line would be re-written as:

import httpx
#....

async def get_stock_price_data(stock):
    print(f"-- stock data:getting stock data for {stock}")
    data = {"stock":stock, "price_open":0, "price_close":0 }

    #*** instead of requests.get('https://finance.yahoo.com/quote/' + stock)) ****
    client = httpx.AsyncClient() 
    stock_page = await client.get( 'https://finance.yahoo.com/quote/' + stock)

    soup = BeautifulSoup(stock_page.text, 'html.parser')
    #<fin-streamer active="" class="Fw(b) Fz(36px) Mb(-4px) D(ib)" data-field="regularMarketPrice" data-pricehint="2" data-symbol="TSLA" data-test="qsp-price" data-trend="none" value="759.63">759.63</fin-streamer>
    data['price_close'] = soup.find('fin-streamer', attrs={"data-symbol":stock, "data-field":"regularMarketPrice"} ).text

    #<td class="Ta(end) Fw(600) Lh(14px)" data-test="OPEN-value">723.25</td>
    data['price_open'] = soup.find( attrs={"data-test":"OPEN-value"}).text
    print(f"-- stock data:done {stock}")
    return data

Ok, that works well. However, but what about the GoogleNews() code. There is no such async version of this function, so how can this be called asynchronously? Well for this, you can actually wrap it around a new thread. A ‘thread’ is way to run a piece of code under the same CPU process but in a parallel. It warrants a whole separate article but for now you can think of it as finding a separate space to execute this independent of the current execution path. However, to execute this in a separate thread, there’s a bit more involved.

The code looks like the following:

### Original Version
def get_recent_news(stock):
    print(f"-- stock news:getting stock data for {stock}")
    gn = GoogleNews()
    search = gn.search(f"stocks {stock}", '24h') #Slow code to run asynchronously
    news = search['entries'][0:3]
    print(f"-- stock news:done {stock}")
    return news

### Asynchronous Version
async def get_recent_news(stock):
    print(f"-- stock news:getting stock data for {stock}")
    gn = GoogleNews()
    search = await asyncio.get_event_loop().run_in_executor( None, gn.search, f"stocks {stock}", '24h')
    news = search['entries'][0:3]
    print(f"-- stock news:done {stock}")
    return news

Here what’s happening is that firstly we are using the await keyword to call the gn.search() function which is now being called through this asyncio.get_event_loop().run_in_executor( .. ) function call. What’s happening here is that we are asking the asyncio module to get access to the event loop (that piece of code that continuously checks for tasks to be done) and then to run in a separate thread. The way it is called is that the parameters must be passed in separate to the function call and hence why the parameters are to be passed in after the function name itself. You will also notice that the whole function can now be defined as async def get_recent_news(stock)

How To Mix Asynchronous And Synchronous Code With Await Async in Python

Now the final problem to be solved is how do we call the two functions of get_stock_price_data( stock ) and get_recent_news(stock) to be run asynchronously, but then wait for both to finish, and THEN run the print. This is where these steps should all be grouped under one function. This is the trick to mix asynchronous and synchronous code.

In order to run a group of tasks in parallel as a group you use asyncio.gather(). However, if you want to execute a synchronous function when ALL tasks that were given to asyncio.gather() is complete, then you should wrap it in another asyncio.gather()

async def process_stock_batch(stock):
    (data, news) = await asyncio.gather( get_stock_price_data( stock ), get_recent_news(stock)  )
    print('-- print:request printing')
    print_stock_update(stock, data, news) 
    print('-- print:done')

async def process_stocks():
    run_stock_list = []
    for stock in stock_list:
        run_stock_list.append(   process_stock_batch(stock) )
    await asyncio.gather( *run_stock_list )

Before we solve it for the real world examples, lets show a simpler example. Suppose we had the following example:

import asyncio, timeit

async def get_web_data_A(index):
    await asyncio.sleep(1)
    print(f"Get Web Data-A[{index}] - sleep 1 second")
        
async def get_web_data_B(index):
    await asyncio.sleep(1)
    print(f"Get Web Data-B[{index}] - sleep 1 second")

async def process(index, start_timer):
    await asyncio.gather( get_web_data_A(index), get_web_data_B(index) )
    print(f"Calculate [{index}] - Elapsed time:[{timeit.default_timer()-start_timer}]")

async def run_all():
    start_timer = timeit.default_timer()
    for index in range(0,2):
        await process(index, start_timer)

if __name__ == '__main__':
    asyncio.run( run_all() )

This has the following output:

What is encouraging with this code, is that even though the call to get_web_data_A() and get_web_data_B() both sleep for 1 second, since they were doing that asynchronously, then the total runtime is still just a little over 1 second. This can be shown by the Calculate [0]... output. However, the problem is that the code still iterates each index sequentially, meaning, that index 0 is processed completely first, and once that’s done, then index 1 is processed. What we want instead is to run all the slow get_web_data_A() and get_web_data_B() first, and then run the code to calculate afterwards. This is where you need to first create the tasks for ALL the iterations, and then call gather() on all the tasks. See the following code:

import asyncio, timeit

async def get_web_data_A(index):
    await asyncio.sleep(1)
    print(f"Get Web Data-A[{index}] - sleep 1 second")
        
async def get_web_data_B(index):
    await asyncio.sleep(1)
    print(f"Get Web Data-B[{index}] - sleep 1 second")

async def process(index, start_timer):
    await asyncio.gather( get_web_data_A(index), get_web_data_B(index) )
    print(f"Calculate [{index}] - Elapsed time:[{timeit.default_timer()-start_timer}]")

async def run_all_2():
    start_timer = timeit.default_timer()
    task_queue = []
    for index in range(0,2):
        task_queue.append( process(index, start_timer) )
    await asyncio.gather( *task_queue )

if __name__ == '__main__':
    asyncio.run( run_all_2() )

Here, in the function async def run_all_2() when we loop, we do not call the blocking code await asyncio.gather... inside the for loop. Instead, we are adding all the tasks to call process(..) into a list called task_queue[], and then at the end of the for loop we are calling await asyncio.gather( *task_queue ) on all tasks in one go. Hence, the output is as follows:

You’ll notice that ALL the get_web_data_A() and get_web_data_B() are being called asynchronously, and then the calculate function is called on all the available data. Hence, the elapsed time for all the iterations is only 1 second, compared to the previous 2 seconds.

So what does this mean for our real world example for getting stock data from Yahoo and then calling Google News asynchronously, and then only printing the data once both are done? Well, the same principle applies. The code is as follows:

import asyncio, httpx, timeit
from bs4 import BeautifulSoup
from pygooglenews import GoogleNews

stock_list = [ "TSLA", "AAPL"]

async def get_stock_price_data(stock):
    print(f"-- stock data:getting stock data for {stock}")
    data = {"stock":stock, "price_open":0, "price_close":0 }

    client = httpx.AsyncClient()
    stock_page = await client.get( 'https://finance.yahoo.com/quote/' + stock)

    soup = BeautifulSoup(stock_page.text, 'html.parser')
    #<fin-streamer active="" class="Fw(b) Fz(36px) Mb(-4px) D(ib)" data-field="regularMarketPrice" data-pricehint="2" data-symbol="TSLA" data-test="qsp-price" data-trend="none" value="759.63">759.63</fin-streamer>
    data['price_close'] = soup.find('fin-streamer', attrs={"data-symbol":stock, "data-field":"regularMarketPrice"} ).text

    #<td class="Ta(end) Fw(600) Lh(14px)" data-test="OPEN-value">723.25</td>
    data['price_open'] = soup.find( attrs={"data-test":"OPEN-value"}).text
    print(f"-- stock data:done {stock}")
    return data

async def get_recent_news(stock):
    print(f"-- stock news:getting stock data for {stock}")
    gn = GoogleNews()
    search = await asyncio.get_event_loop().run_in_executor( None, gn.search, f"stocks {stock}", '24h')
    news = search['entries'][0:3]
    print(f"-- stock news:done {stock}")
    return news

def print_stock_update(stock, data, news):
    print('-- print:starting print')
    print(f"Stock:{ stock }")
    price_change = 0
    if int(float(data['price_open'])) != 0: price_change = round( 100 * ( float( data['price_close'])/float(data['price_open'])-1), 2)
    print(f"Open Price:{data['price_open']} Close Price:{data['price_close']} Change:{price_change}% ")
    print("Latest News:")
    for news_item in news:        
        print( f"{news_item.published}:{news_item.source.title} - {news_item.title}" )

    print("\n")

async def process_stock_batch(stock):
    (data, news) = await asyncio.gather( get_stock_price_data( stock ), get_recent_news(stock)  )
    print('-- print:request printing')
    print_stock_update(stock, data, news) 
    print('-- print:done')

async def process_stocks():
    run_stock_list = []
    for stock in stock_list:
        run_stock_list.append(   process_stock_batch(stock) )
    await asyncio.gather( *run_stock_list )

if __name__ == '__main__':
    start_timer = timeit.default_timer()
    asyncio.run( process_stocks() )
    end_timer = timeit.default_timer()

    print(f"** Total runtime: {end_timer-start_timer} seconds ***")

The key bit of code is in the async def process_stocks() which now iterates over each of the stocks, creates tasks, and then calls await asyncio.gather( *run_stock_list ) on all the stocks in one go, and then in the function process_stock_batch(stock) we have the asynchronous call to (data, news) = await asyncio.gather( get_stock_price_data( stock ), and then the synchronous call to print_stock_update(stock, data, news) once both web data is complete.

Conclusion

The await and async function is an incredibly useful feature of python which takes a bit of getting used to in order to understand the concept, but once you’ve got the hang of it, it can be incredibly useful to get an improve of the performance of your code by leveraging idle time where you are waiting for a task to complete. Remember to be sure about the sequencing and being mindful of whether you care to have a follow-up activity once that task is completed, or you can simply continue to execute.

This not easy to grasp as a beginner, but follow the example code above, and if you get stuck feel free to reach out through our email list below.

How To Get CPU Core Usage with psutil in Python

by Pubs | Aug 16, 2026 | Automation

Intermediate

Getting Per-Core CPU Usage: Quick Example

# quick_cpu_check.py
import psutil
import time

# Pass interval=1 to measure over a 1-second window (recommended)
# percpu=True returns a list -- one value per logical CPU core
core_usage = psutil.cpu_percent(interval=1, percpu=True)

print(f"Logical cores detected: {len(core_usage)}")
print()

for i, pct in enumerate(core_usage):
    bar = "#" * int(pct / 5)
    print(f"  Core {i:>2}: {pct:5.1f}%  [{bar:<20}]")

print()
print(f"  Overall: {psutil.cpu_percent(interval=None):.1f}%")

Output:

Logical cores detected: 8

  Core  0:  23.4%  [####                ]
  Core  1:   8.1%  [#                   ]
  Core  2:  91.3%  [##################  ]
  Core  3:   6.2%  [#                   ]
  Core  4:  12.7%  [##                  ]
  Core  5:   9.4%  [#                   ]
  Core  6:  17.6%  [###                 ]
  Core  7:   5.0%  [#                   ]

  Overall: 21.7%

What is psutil and Why Use It?

Method	Platform	Root Required	Per-Core Data	Python API
psutil	Windows / macOS / Linux	No	Yes	Yes -- clean objects
mpstat	Linux only	No	Yes	Parse subprocess output
top / htop	Unix-like	No	Yes	No -- interactive only
WMI	Windows only	Admin for some	Partial	Via pywin32
/proc/stat	Linux only	No	Yes	Manual file parsing

Install psutil with pip -- it has no dependencies and compiles quickly:

# install_psutil.sh
pip install psutil

Once installed you can import it and immediately start querying system metrics. The sections below walk through each function you need for CPU monitoring.

Logical vs Physical Cores: What cpu_count() Returns

# core_count.py
import psutil

logical = psutil.cpu_count(logical=True)   # includes hyperthreads
physical = psutil.cpu_count(logical=False)  # physical cores only

print(f"Physical cores:  {physical}")
print(f"Logical cores:   {logical}")
print(f"Hyperthreading:  {'Yes' if logical > physical else 'No'}")
print(f"HT ratio:        {logical // physical}x" if physical else "")

Output:

Physical cores:  4
Logical cores:   8
Hyperthreading:  Yes
HT ratio:        2x

Per-Core Frequency with cpu_freq()

# cpu_frequency.py
import psutil

# percpu=True returns a list of scpufreq namedtuples
freqs = psutil.cpu_freq(percpu=True)

if freqs:
    print(f"{'Core':<8} {'Current MHz':>12} {'Min MHz':>10} {'Max MHz':>10}")
    print("-" * 44)
    for i, f in enumerate(freqs):
        print(f"Core {i:<3}  {f.current:>10.0f}   {f.min:>9.0f}  {f.max:>9.0f}")
else:
    # Some Linux VMs do not expose per-core frequency
    overall = psutil.cpu_freq()
    print(f"Per-core freq not available. Overall: {overall.current:.0f} MHz")

Output:

Core     Current MHz    Min MHz    Max MHz
--------------------------------------------
Core 0      3600        800       4200
Core 1      4100        800       4200
Core 2      4200        800       4200
Core 3      3200        800       4200
Core 4      3800        800       4200
Core 5      4000        800       4200
Core 6      4200        800       4200
Core 7      2900        800       4200

Per-Core Time Breakdown with cpu_times()

# cpu_times_breakdown.py
import psutil

times = psutil.cpu_times(percpu=True)

print(f"{'Core':<6} {'User%':>7} {'Sys%':>7} {'Idle%':>7} {'IOWait%':>9}")
print("-" * 40)

for i, t in enumerate(times):
    total = t.user + t.system + t.idle + getattr(t, 'iowait', 0.0)
    if total == 0:
        continue
    user_pct   = t.user   / total * 100
    sys_pct    = t.system / total * 100
    idle_pct   = t.idle   / total * 100
    iowait_pct = getattr(t, 'iowait', 0.0) / total * 100
    print(f"Core {i:<1}  {user_pct:>7.1f} {sys_pct:>7.1f} {idle_pct:>7.1f} {iowait_pct:>9.1f}")

Output:

Core   User%    Sys%   Idle%   IOWait%
----------------------------------------
Core 0   18.2     4.1    77.7       0.0
Core 1    6.5     1.6    91.9       0.0
Core 2   88.4     2.9     8.7       0.0
Core 3    5.1     1.1    93.8       0.0
Core 4   11.3     0.8    87.9       0.1
Core 5    8.7     0.9    90.4       0.0
Core 6   15.2     1.4    83.4       0.0
Core 7    4.2     0.6    95.2       0.0

Memory Monitoring: virtual_memory()

CPU monitoring is rarely useful in isolation -- memory pressure often causes CPU spikes as the OS spends cycles on swapping. Adding memory data to your monitor gives a more complete picture.

# memory_check.py
import psutil

mem = psutil.virtual_memory()
swap = psutil.swap_memory()

def fmt_bytes(n):
    for unit in ('B', 'KB', 'MB', 'GB', 'TB'):
        if n < 1024:
            return f"{n:.1f} {unit}"
        n /= 1024
    return f"{n:.1f} PB"

print("RAM:")
print(f"  Total:     {fmt_bytes(mem.total)}")
print(f"  Available: {fmt_bytes(mem.available)}")
print(f"  Used:      {fmt_bytes(mem.used)}  ({mem.percent:.1f}%)")
print(f"  Buffers:   {fmt_bytes(getattr(mem, 'buffers', 0))}")
print(f"  Cached:    {fmt_bytes(getattr(mem, 'cached', 0))}")
print()
print("Swap:")
print(f"  Total:     {fmt_bytes(swap.total)}")
print(f"  Used:      {fmt_bytes(swap.used)}  ({swap.percent:.1f}%)")

Output:

RAM:
  Total:     15.9 GB
  Available: 9.3 GB
  Used:      6.1 GB  (38.7%)
  Buffers:   312.0 MB
  Cached:    4.2 GB

Swap:
  Total:     2.0 GB
  Used:      0.0 MB  (0.0%)

Threshold Alerting: Raising Warnings When Cores Spike

# cpu_alerts.py
import psutil
import time
import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%H:%M:%S",
)

CPU_WARN_PCT  = 70.0   # warn if any single core exceeds this
CPU_CRIT_PCT  = 90.0   # critical if any core exceeds this
MEM_WARN_PCT  = 80.0   # warn if RAM usage exceeds this
CHECK_INTERVAL = 5     # seconds between checks

def check_once():
    per_core = psutil.cpu_percent(interval=1, percpu=True)
    mem = psutil.virtual_memory()

    for i, pct in enumerate(per_core):
        if pct >= CPU_CRIT_PCT:
            logging.critical("Core %d at %.1f%% -- CRITICAL", i, pct)
        elif pct >= CPU_WARN_PCT:
            logging.warning("Core %d at %.1f%% -- high usage", i, pct)

    if mem.percent >= MEM_WARN_PCT:
        logging.warning("RAM at %.1f%% -- available: %.1f GB",
                        mem.percent, mem.available / 1e9)

if __name__ == "__main__":
    logging.info("Starting CPU/memory monitor (Ctrl+C to stop)")
    try:
        while True:
            check_once()
            time.sleep(CHECK_INTERVAL)
    except KeyboardInterrupt:
        logging.info("Monitor stopped.")

Output:

09:14:01 [INFO] Starting CPU/memory monitor (Ctrl+C to stop)
09:14:02 [WARNING] Core 2 at 73.5% -- high usage
09:14:07 [CRITICAL] Core 2 at 94.1% -- CRITICAL
09:14:12 [CRITICAL] Core 2 at 98.7% -- CRITICAL
09:14:17 [INFO] Monitor stopped.

Real-Life Example: Live Terminal CPU Dashboard

Let us combine everything into a dashboard that refreshes in place every two seconds, showing per-core bars, frequency, and memory -- all in one compact terminal view.

# cpu_dashboard.py
import psutil
import time
import os

CPU_WARN  = 70.0
CPU_CRIT  = 90.0
REFRESH   = 2.0   # seconds between refreshes

def color(pct):
    """Return ANSI color code based on usage percentage."""
    if pct >= CPU_CRIT:
        return "\033[91m"   # bright red
    if pct >= CPU_WARN:
        return "\033[93m"   # yellow
    return "\033[92m"       # green

RESET = "\033[0m"

def make_bar(pct, width=24):
    filled = int(pct / 100 * width)
    return "#" * filled + "-" * (width - filled)

def render():
    os.system("cls" if os.name == "nt" else "clear")
    print("=" * 56)
    print("  psutil CPU Dashboard -- press Ctrl+C to exit")
    print("=" * 56)

    per_core = psutil.cpu_percent(interval=1, percpu=True)
    freqs    = psutil.cpu_freq(percpu=True) or []
    mem      = psutil.virtual_memory()
    logical  = psutil.cpu_count(logical=True)
    physical = psutil.cpu_count(logical=False)

    print(f"  Cores: {physical} physical / {logical} logical\n")

    for i, pct in enumerate(per_core):
        freq_str = ""
        if i < len(freqs):
            freq_str = f"  {freqs[i].current:>5.0f} MHz"
        bar = make_bar(pct)
        c = color(pct)
        print(f"  Core {i:>2}: {c}[{bar}]{RESET} {pct:5.1f}%{freq_str}")

    avg = sum(per_core) / len(per_core) if per_core else 0
    print(f"\n  Avg:    [{make_bar(avg)}] {avg:5.1f}%")
    print()

    mem_bar = make_bar(mem.percent, width=24)
    mc = color(mem.percent)
    avail_gb = mem.available / 1e9
    print(f"  RAM:    {mc}[{mem_bar}]{RESET} {mem.percent:5.1f}%  "
          f"({avail_gb:.1f} GB free)")

    swap = psutil.swap_memory()
    if swap.total > 0:
        swap_bar = make_bar(swap.percent, width=24)
        sc = color(swap.percent)
        print(f"  Swap:   {sc}[{swap_bar}]{RESET} {swap.percent:5.1f}%")

    print()
    print(f"  Updated every {REFRESH}s -- {time.strftime('%H:%M:%S')}")
    print("=" * 56)

if __name__ == "__main__":
    try:
        while True:
            render()
            time.sleep(REFRESH)
    except KeyboardInterrupt:
        print("\nDashboard stopped.")

Output (sample frame):

========================================================
  psutil CPU Dashboard -- press Ctrl+C to exit
========================================================
  Cores: 4 physical / 8 logical

  Core  0: [######------------------]  25.4%   3600 MHz
  Core  1: [#-----------------------]   8.1%   2900 MHz
  Core  2: [######################--]  91.3%   4200 MHz
  Core  3: [#-----------------------]   6.2%   3100 MHz
  Core  4: [###---------------------]  12.7%   3400 MHz
  Core  5: [##----------------------]   9.4%   3200 MHz
  Core  6: [###---------------------]  17.6%   3800 MHz
  Core  7: [#-----------------------]   5.0%   2800 MHz

  Avg:    [####--------------------]  22.0%

  RAM:    [############------------]  51.2%  (7.8 GB free)
  Swap:   [------------------------]   0.0%

  Updated every 2s -- 09:17:44
========================================================

Frequently Asked Questions

Why does cpu_percent() return 0.0 when I call it with no arguments?

When should I use logical=True vs logical=False in cpu_count()?

Does psutil need root/admin privileges?

cpu_freq(percpu=True) returns an empty list on my Linux VM. What is wrong?

Does this code work on Windows without any changes?

Can I get CPU temperature with psutil?

Conclusion

« Older Entries

How to Build a Twitter Bot with Python and Twitter API v2

Easy guide for data storage options in Python

Further Reading: For more details, see the Python asyncio documentation.

Frequently Asked Questions

What is async/await in Python?

async def defines a coroutine function and await pauses execution until an asynchronous operation completes. This enables concurrent I/O operations without threading, using the asyncio event loop.

When should I use async/await instead of threading?

Use async/await for I/O-bound tasks like network requests and database queries with many concurrent connections. Use threading for CPU-bound tasks or libraries that do not support async.

How do I run multiple async tasks concurrently?

Use asyncio.gather(task1(), task2()) to run multiple coroutines concurrently. Use asyncio.create_task() to schedule without immediately waiting.

What does ‘coroutine was never awaited’ mean?

You called an async function without await. Async functions return coroutine objects that must be awaited. Add await before the call or use asyncio.run() from synchronous code.

Can I mix synchronous and asynchronous code?

Yes. Use asyncio.run() to call async from sync. Use loop.run_in_executor() to run blocking functions inside async code without blocking the event loop.

Continue Learning Python

Tutorials you might also find useful:

Post Views: 1,195

Python Await Async Tutorial with Real Examples and Simple Explanations

What is async await in Python?

How does async await work in Python?

Cooking Hamburgers Asynchronously and coding the event loop manually

How To Get CPU Core Usage with psutil in Python

Getting Per-Core CPU Usage: Quick Example

What is psutil and Why Use It?

Logical vs Physical Cores: What cpu_count() Returns

Per-Core Frequency with cpu_freq()

Per-Core Time Breakdown with cpu_times()

Memory Monitoring: virtual_memory()

Threshold Alerting: Raising Warnings When Cores Spike

Real-Life Example: Live Terminal CPU Dashboard

Frequently Asked Questions

Why does cpu_percent() return 0.0 when I call it with no arguments?

When should I use logical=True vs logical=False in cpu_count()?

Does psutil need root/admin privileges?

cpu_freq(percpu=True) returns an empty list on my Linux VM. What is wrong?

Does this code work on Windows without any changes?

Can I get CPU temperature with psutil?

Conclusion

Related Articles

Async Await Code Example in Python

Async Asynchronous Calling Another Async Function Code Example

Async Await Real World Example With Web Crawler in Python

How To Mix Asynchronous And Synchronous Code With Await Async in Python

Conclusion

How To Get CPU Core Usage with psutil in Python

Getting Per-Core CPU Usage: Quick Example

What is psutil and Why Use It?

Logical vs Physical Cores: What cpu_count() Returns

Per-Core Frequency with cpu_freq()

Per-Core Time Breakdown with cpu_times()

Memory Monitoring: virtual_memory()

Threshold Alerting: Raising Warnings When Cores Spike

Real-Life Example: Live Terminal CPU Dashboard

Frequently Asked Questions

Why does cpu_percent() return 0.0 when I call it with no arguments?

When should I use logical=True vs logical=False in cpu_count()?

Does psutil need root/admin privileges?

cpu_freq(percpu=True) returns an empty list on my Linux VM. What is wrong?

Does this code work on Windows without any changes?

Can I get CPU temperature with psutil?

Conclusion

Related Articles

Related Articles

Frequently Asked Questions

What is async/await in Python?

When should I use async/await instead of threading?

How do I run multiple async tasks concurrently?

What does ‘coroutine was never awaited’ mean?

Can I mix synchronous and asynchronous code?

Continue Learning Python

Related Articles

Submit a Comment Cancel reply