How To Lock A Class in Python

Last Updated on September 12, 2022

You can lock a class using a mutex lock via the threading.Lock class.

In this tutorial you will discover how to lock a class in Python.

Let’s get started.

Table of Contents

Need to Lock a Class

A thread is a thread of execution in a computer program.

Every Python program has at least one thread of execution called the main thread. Both processes and threads are created and managed by the underlying operating system.

Sometimes we may need to create additional threads in our program in order to execute code concurrently.

Python provides the ability to create and manage new threads via the threading module and the threading.Thread class.

You can learn more about Python threads in the guide:

Threading in Python: The Complete Guide

In concurrent programming, we may need to lock a class to protect it from being called concurrently from multiple threads.

How can we lock a class in Python?

Run loops using all CPUs, download your FREE book to learn how.

How to Lock a Class

We can lock a class using a mutual exclusion (mutex) lock.

Mutex Lock

A mutual exclusion lock or mutex lock is a synchronization primitive intended to prevent a race condition.

A mutex lock can be used to ensure that only one thread at a time executes a critical section of code at a time, while all other threads trying to execute the same code must wait until the currently executing thread is finished with the critical section and releases the lock.

Python provides a mutual exclusion lock via the threading.Lock class.

First, an instance of the lock must be created.

...

# create a lock

lock = Lock()

The lock can then be used to protect critical sections of code that require serialized or mutually exclusive access.

This can be achieved by first acquiring the lock by calling the acquire() function before accessing a critical section, then releasing the lock by calling release() after the critical section.

For example:

...

# acquire the lock

lock.acquire()

# critical section

# ...

# release the lock

lock.release()

Only one thread can have the lock at any time.

If the lock is being held, another thread attempting to acquire the lock will block until the lock is available.

We can also use the lock via the context manager interface.

This will allow the critical section to be a block within the usage of the lock and for the lock to be released automatically once the block has completed.

For example:

...

# acquire the lock

with lock:

# ...

The context manager is the preferred approach to using the lock as it guarantees the lock is always released, even if an unexpected Error or Exception is raised.

You can learn more about mutex locks in the tutorial:

How to Use a Mutex Lock in Python

Now that we know about mutex locks, let’s look at how to lock a class.

Lock a Class

A class can be locked using a mutex lock.

Recall that a class defines an object that may have instance variables and instance methods.

A class may also have class variables and class methods defined with the @classmethod decorator. It may also have static variables and static methods defined with the @staticmethod decorator.

We will look at locking a class via locking class methods instead of static methods, designed to maintain class variables state in the class across all object instances.

This can be achieved by creating an instance of a lock as a class variable.

We may have a custom class named CustomClass. We can define a mutex lock as a class variable and initialize it prior to the class constructor.

For example:

# custom class

class CustomClass:

# initialize mutex lock as a class variable

lock = threading.Lock()

The lock class variable can then be used within each class method of the class.

We may acquire the lock as the first line of any class method in the class, and release it as the last line of the method.

This can be achieved using the context manager.

Recall, we can use the @classmethod decorator to define a class method.

For example:

# custom class

class CustomClass:

# initialize mutex lock as a static variable

lock = threading.Lock()

# a method to do something

@classmethod

def work(cls):

# acquire the lock

with cls.lock:

# ...

This shows how we might lock a class to protect it from race conditions.

Next, let’s look at a worked example.

Download Now: Free Threading PDF Cheat Sheet

Example of Locking a Class

We can develop an example of locking a class.

In this example we can define a new class that maintains a counter class variable. We will call a class method to increment the counter and another class method to print the value of the counter. All access to the counter is protected by a mutex lock.

We can then increment the counter from many threads and the lock will protect changes to the counter from race conditions.

Recall that a class variable is accessible on the class itself and by all instances of the class. Similarly, class methods are also accessible on the class itself, and all instances of the class. They are different from instance variables and instance methods that are only accessible on a particular instance of the class.

Incrementing a counter is a good functionality to protect from race conditions as it is inherently not thread safe. You can learn more about this in the tutorials:

First, we can define a class with the counter class variable and our custom methods.

We will call the class ThreadSafeCounter.

# thread-safe counter class

class ThreadSafeCounter:

# ...

We can initialize the counter and the lock protecting the counter as class variables directly on the class definition.

...

# initialize mutex lock as a class variable

lock = Lock()

# initialize the counter as a class variable

counter = 0

We can then define an increment() class method that will first acquire the lock, and will then increment the counter value.

Recall that class methods can be defined using the @classmethod decorator prior to the definition of the method and take the class as the first argument. The class argument can then be used to access the class variables such as the count and the lock.

# increment the counter

@classmethod

def increment(cls):

# acquire the lock

with cls.lock:

# increment the counter

cls.counter += 1

Finally, we can define a report() class method that acquires the lock, then prints the current value of the counter.

# report the counter value

@classmethod

def report(cls):

# acquire the lock

with cls.lock:

# report the counter value

print(cls.counter)

Tying this together, the complete ThreadSafeCounter class is listed below.

# thread-safe counter class

class ThreadSafeCounter:

# initialize mutex lock as a class variable

lock = Lock()

# initialize the counter as a class variable

counter = 0

# increment the counter

@classmethod

def increment(cls):

# acquire the lock

with cls.lock:

# increment the counter

cls.counter += 1

# report the counter value

@classmethod

def report(cls):

# acquire the lock

with cls.lock:

# report the counter value

print(cls.counter)

Next, we can define a target task function to be executed in new threads.

This function will not take any arguments. It will then loop for a large number of times, e.g. 10,000, and call the increment() class method each iteration.

The task() function below implements this.

# task function to increment the counter in a new thread

def task():

for i in range(10000):

ThreadSafeCounter.increment()

In the main thread, we can create a large number of threads that will all be configured to execute our task() function on the same ThreadSafeCounter class.

This can be achieved in a list comprehension where we will create a list of 1,000 configured threads.

...

# create many threads

threads = [Thread(target=task) for _ in range(1000)]

We can then iterate over the list of threads and start them each in turn.

...

# start threads

for thread in threads:

thread.start()

The main thread can then block until all of the threads have finished.

...

# wait for threads

for thread in threads:

thread.join()

Finally, the value of the counter can be reported.

With 1,000 threads each incrementing the counter 10,000 times, we expect the final counter value to be 10,000,000 (ten million), given no race condition updating the counter.

...

# report the counter value

counter.report()

Tying this together, the complete example is listed below.

# SuperFastPython.com

# example of locking a class

from threading import Thread

from threading import Lock

# thread-safe counter class

class ThreadSafeCounter:

# initialize mutex lock as a class variable

lock = Lock()

# initialize the counter as a class variable

counter = 0

# increment the counter

@classmethod

def increment(cls):

# acquire the lock

with cls.lock:

# increment the counter

cls.counter += 1

# report the counter value

@classmethod

def report(cls):

# acquire the lock

with cls.lock:

# report the counter value

print(cls.counter)

# task function to increment the counter in a new thread

def task():

for i in range(10000):

ThreadSafeCounter.increment()

# create many threads

threads = [Thread(target=task) for _ in range(1000)]

# start threads

for thread in threads:

thread.start()

# wait for threads

for thread in threads:

thread.join()

# report the counter value

ThreadSafeCounter.report()

Running the example first creates 1,000 threads that are configured to execute our task() function.

The threads are then started and the main thread blocks until the threads finish.

Each thread loops for 10,000 iterations and each iteration it calls the increment() class method on the ThreadSafeCounter class directly.

Each call to the increment() class method first acquires the class variable lock, then updates the value of the counter. If the lock has already been acquired, other threads will block and wait for the lock to become available. This ensures that all updates to the counter are mutually exclusive and serialized (one at a time).

All threads finish, then the final value of the counter is reported.

We can see that the expected value of ten million is reported, and will be reported every time the program is run due to the protection provided by locking the class itself.

10000000

Free Python Threading Course

Download your FREE threading PDF cheat sheet and get BONUS access to my free 7-day crash course on the threading API.

Discover how to use the Python threading module including how to create and start new threads and how to use a mutex locks and semaphores

Learn more