Forem: Konstantinos Blatsoukas

python daily #1 (pattern matching)

Konstantinos Blatsoukas — Wed, 09 Jul 2025 12:00:40 +0000

don't try to match a constant (there will be a match, but not what you would expect)

example:

ERROR = "error"
SUCCESS = "success"

error_obj = {"bar_status": "error"}

match error_obj:
    case {"bar_status": ERROR}:
        print(f"error status matched! status: {ERROR}")
    case _:
        print("No match found")

if you execute the above code, the output will be:

error status matched! status: error

this was my intention, seems that works as expected (but wait for the next one...)

ERROR = "error"
SUCCESS = "success"

error_obj = {"bar_status": "error"}

match error_obj:
    case {"bar_status": SUCCESS}:
        print(f"error status matched! status: {SUCCESS}")
    case _:
        print("No match found")

print(f"SUCCESS status: {SUCCESS}")

output:

error status matched! status: error
SUCCESS status: error

again the same output! what happened?

According to PEP 636 the SUCCESS will be interpreted as a capture pattern. Which means, that is going to match any value, that "bar_status" key has, and use it as a SUCCESS variable (also the SUCCESS variable has now the value error).

The solution for matching constants is to use enums.

The above match could be written as:

class Status(enum.StrEnum):
    ERROR = "error"
    SUCCESS = "success"


error_obj = {"bar_status": "success"}

match error_obj:
    case {"bar_status": Status.ERROR}:
        print(f"error status matched! status: {Status.ERROR}")
    case {"bar_status": Status.SUCCESS}:
        print(f"success status matched! status: {Status.SUCCESS}")
    case _:
        print("No match found")

Cheers!

frictionless, first transform and then validate

Konstantinos Blatsoukas — Mon, 26 May 2025 13:23:29 +0000

In a recent task, I had to use the frictionless library for validating a csv file, after a transformation. So, the weird part in this task was that, the transformation step comes first, and then the validation on the transformed data. Usually, a validation is performed first and then the data transformation (if any required).

Let's see, how I did that:

from frictionless import Resource, Schema, transform, Pipeline, steps, validate

# you can find the pipeline implementation in the next code section
    build_successfully, trx_pipeline, error_messages = build_trx_pipeline(read_file, schema.field_names)
if not build_successfully:
    error_file_name = file.filename
    error_report[error_file_name] = error_messages
        continue

# create a Pipeline object based on a list of Steps, see details below
trx_pipeline = Pipeline(steps=build_trx_pipeline)

# create a resource, based on the original file (no schema was specified)
original_resource = Resource(data=read_file, format="csv")

# apply the trandormation to the original file
transformed_resource = transform(original_resource, pipeline=trx_pipeline)

# explicitly specify  the validation schema
transformed_resource.schema = Schema("/my_schema.schema.json")

transform the data to the desired schema (based on the json schema)


# read file, I am using the fastAPI framework
read_file = await file.read()

# build the pipeline transformation, in term of steps
def build_trx_pipeline(read_file, schema_field_names) -> tuple[bool, list[Any], list[str]]:

    csv_headers = str(read_file.splitlines()[0].decode("utf-8")).split(",")

    pipeline_steps = []

    # keep only the rows with apprtype == 'TRN' and tank == 'MAIN'
    pipeline_steps.append(steps.row_filter(formula="foo == 'X' and bar == 'HELLO'"))

    pipeline_steps.append(steps.field_remove(names=list({"a", "b"})))

    # rename fields to match the lab dga SAI schema
    renaming_steps = [
            steps.field_update(name=a_name, descriptor={"name": b_name})
            for a_name, b_name in toa2sai_fields.items()
            if a_name in csv_headers
      ]
    pipeline_steps.extend(renaming_steps)

    #    .....

    return True, pipeline_steps, []

another uv post :)

Konstantinos Blatsoukas — Wed, 22 Jan 2025 12:16:08 +0000

installing uv

For installing uv, only curl tool required.

curl -LsSf https://astral.sh/uv/install.sh | sh

uv site

starting a project

uv init example

## if you want to specify a python version, you can that, using --python option
uv init example --python 3.11

this produces a project under a folder name example

example/
├── .gitignore
├── .python-version
├── README.md
├── hello.py
└── pyproject.toml

In the toml file the the required python version is specified (also in the .python-version file).
Also, in the same file the project dependencies are specified, as well.

running a project

uv run hello.py
# program output
Using CPython 3.11.11
Creating virtual environment at: .venv
Hello from example!

Python 3.11.11 wasn't installed on my local system.
uv downloaded the specified version and also created a virtual environment for us.

Let's examine the folder structure, after the

uv run

command.

.
├── .python-version # default python version
├── .venv
├── README.md
├── hello.py
├── pyproject.toml
└── uv.lock

A .venv was created for us, automatically! and uv.lock file (which includes the dependency details, more bellow)

managing dependencies

You can add/remove projects dependencies in pyproject.toml file, on dependencies list:

[project]
name = "backend"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
    "aio-pika==9.5.0",
    "aiormq==6.8.1",
]

In the above example, we added two dependencies aio-pika and aiormq.

After the toml editing, a sync is required (sync updates also the lock file, venv etc.)

uv sync

The other way is to use the following commands:

uv add 'aio-pika==9.5.0'
uv add 'aiormq==6.8.1'

(Imagine that the dependencies were empty) The above commands will add both libs in the dependency list (toml file), update the venv, and the lock file.
Moreover, uv will download and install all the required transitive dependencies.

More info about maanging dependencies, you can find here:

uv managing dependencies

deps using pip vs uv

This is a simple test for comparing how much time pip install takes to install libs vs the uv.

Imagine that our current project needs the following libs, to be installed:

aio_pika
asyncpg
boto3
datadog
fastapi
geojson
httpx
loguru
pydantic_settings
pypika
python-jose
python-multipart
uvicorn

Let's try doing this using pip and using uv.

using pip

python -m venv vevn
source vevn/bin/activate
pip install pip-tools
pip-compile requirements.in
pip install --no-cache-dir -r requirements.txt

using time in front of the

pip install

I got the following time durations:

real    0m16.852s
user    0m9.211s
sys     0m0.811s

using uv

All the above libs are included on toml file dependencies list:

uv sync --no-cache

using time in front of the

uv syn --no-cache

, I got the following time durations:

real    0m2.553s
user    0m1.511s
sys     0m1.399s

comparing the real time, we get a X6.5 reduction time!

changing python version for a project

uv python pin 3.13 # no need for python 3.13 to be present (if not, it will be installed automatically)
Updated `.python-version` from `3.11` -> `3.13`

let's try now to run again our project

Using CPython 3.13.1
Removed virtual environment at: .venv
Creating virtual environment at: .venv
Hello from example!

CPython 3.13.1, downloaded and installed.
Moreover, previous vevnv automatically deleted and a new one was created.

not sure if I want to migrate to uv

If you are not sure of using uv, you can get some of the benefits, by just putting


 word in front of any pip command that you have:



```sh
# use this:
uv pip compile requirements.in -o requirements.txt
# instead of this
# pip-compile requirements.in -o requirements.txt

You will get the dependencies much faster.

python management

installing the latest version

uv install python

installing specific version

uv install python 3.12

installing multiple python versions

uv install python 3.11 3.12

viewing python versions

uv python list

#output
cpython-3.14.0a4+freethreaded-linux-x86_64-gnu    <download available>
cpython-3.14.0a4-linux-x86_64-gnu                 <download available>
cpython-3.13.1+freethreaded-linux-x86_64-gnu      <download available>
cpython-3.13.1-linux-x86_64-gnu                   <download available>
cpython-3.12.8-linux-x86_64-gnu                   /home/kostas/.local/share/uv/python/cpython-3.12.8-linux-x86_64-gnu/bin/python3.12
cpython-3.11.11-linux-x86_64-gnu                  /home/kostas/.local/share/uv/python/cpython-3.11.11-linux-x86_64-gnu/bin/python3.11
cpython-3.10.16-linux-x86_64-gnu                  <download available>
cpython-3.10.12-linux-x86_64-gnu                  /usr/bin/python3.10
cpython-3.10.12-linux-x86_64-gnu                  /usr/bin/python3 -> python3.10
cpython-3.10.12-linux-x86_64-gnu                  /usr/bin/python -> /usr/bin/python3
cpython-3.10.12-linux-x86_64-gnu                  /bin/python3.10
cpython-3.10.12-linux-x86_64-gnu                  /bin/python3 -> python3.10
cpython-3.10.12-linux-x86_64-gnu                  /bin/python -> /usr/bin/python3
cpython-3.9.21-linux-x86_64-gnu                   <download available>
cpython-3.8.20-linux-x86_64-gnu                   <download available>
cpython-3.7.9-linux-x86_64-gnu                    <download available>
pypy-3.10.14-linux-x86_64-gnu                     <download available>
pypy-3.9.19-linux-x86_64-gnu                      <download available>
pypy-3.8.16-linux-x86_64-gnu                      <download available>
pypy-3.7.13-linux-x86_64-gnu                      <download available>

You can see the already installed python versions, along the path that each version is installed.
If the python version you specified in the command is already present, this version will be used for executing your python program.

Kubernetes pods

Konstantinos Blatsoukas — Wed, 30 Oct 2024 10:38:58 +0000



Pod status:               kubectl get pods
Pod deletion:             kubectl delete pods/kuard
                          kubectl delete -f kuard-pod.yaml
Run a pod  (manifest):    kubectl apply -f kuard-pod.yaml
Listing pods:             kubectl get pods # -o yaml or -o wide
Pod details:              kubectl describe pods <pod-name>
port-forwarding:          kubectl port-forward <pod-name> 8080:8080
Pod logs:                 kubectl logs <pod-name> # with -f you can stream logs, 
                          #--previous if the pods are restarting
executing cmds:           kubectl exec <pod-name> date
                          kubectl exec -it <pod-name> ash
copying files:            kubectl cp <pod-name>:/captures/capture3.txt ./capture3.txt

AWS CloudWatch notes

Konstantinos Blatsoukas — Fri, 18 Oct 2024 08:31:34 +0000

AWS Monitor (AmazonCloud watch)

Data points generted from resources --> Metrics/Time = Statistics

AmazonCloud watch alarms

You can define a threshold, and if this threshold is reached an alarm can be fired.

Possible alarm states:

alarm (outside the threshold)
ok (withine the threshold)
insufficient data (alarm just started)

Monitoring types:

basic monitoring (every 5'), free
detailed monitoring (every 1')

Metrics in CloudWatch

Namespaces: a container for organizing related metrics.
Metrics in different namespaces, are completely isolated.
Dimensions: a name/valeu pair that CloudWatch attaches to each metric.
Usefull, for filtering metrics.

CloudWatch Logs

Log event: a log event recorder by some resource, with timespam and event message
Log stream: Log events grouped into log streams.
Log groups: log streams are organized in log groups. - log groups are sharing the following properties: - Retention period - premission settings

AWS storage notes

Konstantinos Blatsoukas — Thu, 17 Oct 2024 09:53:22 +0000

AWS storage

Storage types: Block storage or Object storage

For frequent modifications, block storage is the proffered option

Amazon EBS (Elastic Block Store)

SSD or HDD backed volumes

Amazon RDS is an EBS (relational databases)

(-) attached to a specific EC2

(-) size limitations

(+) scalable, available

(+) snapshots (backups)

Amazon S3

is an object storage
not tied to an EC2
Amazon RDS, relational DBs, all mainstream are supported (MySQL, postgres etc.)
Amazon Dynamo DB (NoSQL, key-value)
Amazon DocumentDB (CMS, catalogs et.)
Amazon Nepture (graphs)
Amazon QLDB (immutable ledgers etc.)

Database Type | Use Cases | AWS Service

Relational | Traditional applications, ERP, CRM, e-commerce | Amazon RDS, Amazon Aurora, Amazon Redshift
Key-value | High-traffic web apps, e-commerce systems, gaming applications | Amazon DynamoDB
In-memory | Caching, session management, gaming leaderboards, geospatial applications | Amazon ElastiCache for Memcached, Amazon ElastiCache for Redis
Document | Content management, catalogs, user profiles | Amazon DocumentDB (with MongoDB compatibility)
Wide column | High-scale industrial apps for equipment maintenance, fleet management, and route optimization | Amazon Keyspaces (for Apache Cassandra)
Graph | Fraud detection, social networking, recommendation engines | Amazon Neptune
Time series | IoT applications, DevOps, industrial telemetry | Amazon Timestream
Ledger | Systems of record, supply chain, registrations, banking transactions | Amazon QLDB

intro shell bash scripting

Konstantinos Blatsoukas — Wed, 16 Oct 2024 12:35:00 +0000

writting a simple script

#~/bin/bash
cd /etc
cat ./passwd

give execution permissions

chmod +x test_script.sh

pass data

#~/bin/bash
echo good morning $1 I hope you haveiong a great day

Environmental variable

# craete an environmental variable
export USERNAME=kostas

#~/bin/bash
echo good morning $USERNAME I hope you haveiong a great day

zip a file and move it to a new directory

#!/bin/bash

source=$1
dest=$2

tar -cvf backup.tar $source
gzip backup.tar
mv backup.tar.gz $dest

mkdir audiofiles
cd audiofiles/
touch f1 f2 f3
cd ..

READ and TEST commands, if conditionals

#!/bin/bash

# read user input
read -p "enter your desired username " username
echo you have chosen $username as your username

$?
# 0 success, 1 and above error


#!/bin/bash
if [ 5 -gt 3 ]; then
    echo yes it is greater than 5
else
    echo no it is not
fi

loops

#!/bin/bash

for name in natthew mark luke
do
    echo good morning $name
done

loops using a bash command

#!/bin/bash
count=1
for file in `ls *.log`
do
    echo "log file number $count: $file"
    count=$((count+1))
done

loops, the while case

#!/bin/bash
while true
do
    touch file-`date +%s`.txt
    sleep 3
done

loops, the until case

#!/bin/bash
until `ssh 192.168.1.100`
do
    echo "still attemting..."
done

functions in Bash

mkdir newdir
cd newdir
touch file1 file2 file3 file4

#!/bin/bash

source=$1
dest=$2

function fileExists() {
    # check file existence
    if [ -f  $dest/backup3.tar.gz ]; then
        echo this file already exists
        exit
    fi
}


function compressDir() {
tar -cvf backup3.tar $source
gzip backup3.tar
mv backup3.tar.gz $dest
}

fileExists
compressDir

venv

Konstantinos Blatsoukas — Tue, 15 Oct 2024 07:10:46 +0000

python vevnv

Is used for separating packages versions per project.
For example, one project may use a package named foo-1-0, but another project uses foo-2-0.
We can not separate them using the global environment.
The solution is, to create two different virtual environemnts.

list all global packages

pip list # list all packages

create a new virtual environment

# a new virtual environment with name project_env will be craeted
python -m venv project_env

activate venv

activate

# activate on ubuntu
source myenv/bin/activate

deactivate venv

deactivate

generate packages version for requirements.txt

pip freeze
pip install -r requirements.txt

generate required python packages

python -m venv vevn
source vevn/bin/activate
pip install pip-tools
pip-compile requirements.in
pip install -r requirements.txt

Debezium integration

Konstantinos Blatsoukas — Fri, 16 Aug 2024 08:23:23 +0000

Design and implementation using RabbitMQ and Debezium

Using kafka as a queue, and debezium connector (part of kafka connect) for monitoring the database, the following components are used:

Debezium Server: Standalone software that monitors a database for changes, and sends the changes to a queue system (like RabbitMQ).

RabbitMQ Broker: queuing system that keeps the database changes events into queues.

RabbitMQ Consumer: consumes the database events from a RabbitMQ queue

The guide (using RabbitMQ as a queue broker) is divided into three parts:

High-Level Design: A brief description of each component's functionality and how they are interconnected.

Implementation with Docker Compose: A step-by-step guide to setting up the components using a Docker Compose file.

RabbitMQ Consumer Integration: How to integrate the RabbitMQ Consumer with the Sapient app.

High level design

The high level design, is depicted in the following image:

The main components of the system are:

Debezium Server: monitors the PostgrSQL for databsae changes. If any change occurs (CREATE, DELETE, and UPDATE, not the structural ones) an event is send in the RabbitMQ broker

RabbitMQ Broker: stores the database changes in multiple queues. In our case, because we need each change to be broadasted to all queues, the exchange type is set to fanout.

RabbitMQ consumer integration to Sapient app: Consumes the database changes from a queue. Each queue is related to each app instance and is bound to a signle exchange.

Implementation through a Docker Compose file

create a folder named debezium_conf. On this folder, create a file named application.properties.

The file contains the Debezium Server configuration. Two crucial parts of the configuration are:

source configuration (the database that we want to monitor, postgreSQL in our case)

sink configuration (the queue system that we want to send the changes, RabbitMQ in our case)

the contents of application.properties file:



# Sink connector config - RabbitMQ
debezium.sink.type=rabbitmq
debezium.sink.rabbitmq.connection.host=rabbitmq
debezium.sink.rabbitmq.connection.port=5672
debezium.sink.rabbitmq.connection.username=guest
debezium.sink.rabbitmq.connection.password=guest
debezium.sink.rabbitmq.ackTimeout=3000
debezium.sink.rabbitmq.exchange=db_changes

# Source connector config - PostgreSQL
debezium.source.connector.class=io.debezium.connector.postgresql.PostgresConnector
debezium.source.database.dbname=sapient_insights
debezium.source.database.hostname=database
debezium.source.database.password=camlin
debezium.source.database.port=5432
debezium.source.database.user=camlin
debezium.source.offset.flush.interval.ms=0
debezium.source.offset.storage.file.filename=data/offsets.dat
debezium.source.plugin.name=pgoutput
debezium.source.topic.prefix=tutorial
debezium.source.table.include.list=public.preferences
a sour
# Key Converter configuration
debezium.source.key.converter=org.apache.kafka.connect.json.JsonConverter
debezium.source.key.converter.schemas.enable=false

# Value Converter configuration
debezium.source.value.converter=org.apache.kafka.connect.json.JsonConverter
debezium.source.value.converter.schemas.enable=false

# Format config
debezium.format.key=json
debezium.format.value=json

# Quarkus
quarkus.log.console.json=false

# SMT configuration
debezium.transforms=flatten
debezium.transforms.flatten.type=org.apache.kafka.connect.transforms.Flatten$Value
debezium.transforms.flatten.delimiter=.

add debezium server to docker compose file:



services:
  ...
    debezium:
      image: debezium/server:2.7
      healthcheck:
        test: curl http://debezium:8080/q/health || exit 1
        interval: 5s
        timeout: 5s
        retries: 5
      depends_on:
        - database
        - rabbitmq
      restart: on-failure
      volumes:
        - ./debezium_conf:/debezium/conf:readonly
      ports:
        - "8080:8080"
  ...

create a folder named rabbitmq_conf. On this folder, create three files:

definitions.json: defines an exchange, a user, and the permissions for this user. The excahnge is fanout type:



{
  "rabbit_version": "3.13.1",
  "rabbitmq_version": "3.13.1",
  "product_name": "RabbitMQ",
  "product_version": "3.13.1",
  "vhosts": [
    {
      "name": "/"
    }
  ],
  "users": [
    {
      "name": "guest",
      "password_hash": "9/1i+jKFRpbTRV1PtRnzFFYibT3cEpP92JeZ8YKGtflf4e/u",
      "tags": ["administrator"]
    }
  ],
  "permissions": [
    {
      "user": "guest",
      "vhost": "/",
      "configure": ".*",
      "read": ".*",
      "write": ".*"
    }
  ],
  "exchanges": [
    {
      "name": "db_changes",
      "vhost": "/",
      "type": "fanout",
      "durable": true,
      "auto_delete": false,
      "internal": false,
      "arguments": {}
    }
  ]
}

rabbitmq.conf, defines directory in which the above definitions are loaded. Moreover, a default user, and the listening ports are set.



default_user = guest
default_pass = guest

listeners.tcp.default = 5672
management.tcp.port = 15672
management.load_definitions = /etc/rabbitmq/definitions.json

Dockerfile: the dockerfile from wich RabbitMQ is created (takes into account the definitions and the rabbitMQ configuration).



FROM rabbitmq:3-management

COPY rabbitmq.conf /etc/rabbitmq
COPY definitions.json /etc/rabbitmq

add RabbitMQ to docker compose file:



services:
  ....
  rabbitmq:
    build:
      context: ./rabbitmq_conf/
      dockerfile: Dockerfile
    image: rabbitmq-demo
    container_name: rabbitmq
    ports:
      - "5672:5672"
      - "15672:15672"
      - "25672:25672"
  ....

RabbitMQ Consumer integration

We need a mechanism for receiving the database change notifications from the RabbitMQ Broker in an asynchronous way (non-blocking). This is possible through a RabbitMQ Consumer. The aiopika, provides an asynchronous solution for performing this, more information about aiopika, see this link.

add the following lines:



import backoff
import json
import uuid

from aio_pika import connect_robust, IncomingMessage
from loguru import logger

from .websocket import manager
from . import processor
from ..config import config


RABBIT_URL_CONNECTION = (
    f"amqp://{config.RABBIT_USER_NAME}:{config.RABBIT_USER_NAME}@{config.RABBIT_HOST}/"
)


@backoff.on_exception(
    backoff.expo,
    Exception,
    max_tries=7,
    jitter=backoff.full_jitter,
    on_backoff=lambda details: logger.warning(
        f"Retrying to connect to RabbitMQ: attempt {details['tries']}"
    ),
)
async def start_rabbit_consumer():
    connection = await connect_robust(RABBIT_URL_CONNECTION)
    channel = await connection.channel()

    queue = await channel.declare_queue(str(uuid.uuid4()), durable=True)
    await queue.bind(config.RABBIT_EXCHANGE)

    await queue.consume(process_message, no_ack=False)


async def start_consumer():
    try:
        await start_rabbit_consumer()
    except Exception as e:
        logger.error(f"Failed to connect to RabbitMQ: {e}")


async def process_message(message: IncomingMessage):
    async with message.process():
        message_dict = json.loads(message.body.decode("utf-8"))
        transformed_message = processor.transform(None, message_dict)
        await manager.send_message(json.dumps(transformed_message))

# Python dictionary

Konstantinos Blatsoukas — Thu, 01 Aug 2024 08:43:42 +0000

Mapping methods

>>> from collections import abc
>>> my_dict = {}
>>> isinstance(my_dict, abc.Mapping)
True

When an object considered hashbale?

In order an aobject to be hashable needs to have a hash value that never changes during its lifetime.
Implementation of:
-

        __hash()__
    ```
{% endraw %}

    -
{% raw %}
 ```python
        __eq()__
    ```
{% endraw %}

are mandatory.

A tuple is hashable, if all items are hashable, example:
{% raw %}


```python
>>> tt = (1, 2, (30, 40))
>>> hash(tt)
8027212646858338501
>>> tl = (1, 2, [30, 40])
>>> hash(tl)
Traceback (most recent call last):
 File "<stdin>", line 1, in <module>
TypeError: unhashable type: 'list'
>>> tf = (1, 2, frozenset([30, 40]))
>>> hash(tf)
-4118419923444501110

dict comprehensions

>>> countries_codes = [('DK', 'Denmark'), ('GR', 'Greece'), ('IT', 'Italy')]
>>> codes_dict = {code:  country_name for code, country_name in  countries_codes}
>>> codes_dict
{'DK': 'Denmark', 'GR': 'Greece', 'IT': 'Italy'}

setdefault

my_dict.setdefault(key, []).append(new_value)

===

# …is the same as running…
if key not in my_dict:
 my_dict[key] = []
my_dict[key].append(new_value)

default dict

>>> from collections import defaultdict
>>> my_dict = defaultdict(list) 
>>> my_dict['a'].append(3) 
>>> my_dict['a'] 
[3]

Immutable map

>>> from types import MappingProxyType
>>> d = {1: 'A'}
>>> d_proxy = MappingProxyType(d)

Set

# union
a | b
# intersection
a & b
# difference
a - b

Python string methods examples

Konstantinos Blatsoukas — Fri, 22 Mar 2024 12:41:13 +0000

Here are some useful python string methods examples:

transform a sentece to title:

print("My python text".title())
# output
# My Python Text

check the ending of a phrase, with endsWith():

# returns a boolean in case that the sentence ends with the
# provided parameter

print("My python text".endswith("."))
# False
print("My python text.".endswith("."))
# True

remove any tabs, whitespaces from a given string (from the beginning and the end)

# returns a boolean in case that the sentence ends with the
# provided parameter

print("   My python text.   ".strip())
# My python text.

find and replace words or string sequences:

print("I bake good cakes.".replace("good", "amazing"))
# I bake amazing cakes.

Vscode black formatter

Konstantinos Blatsoukas — Fri, 22 Mar 2024 10:57:36 +0000

Assumptions:

python extension is installed on vscode python extension
balck formatter extension is installed on vscode balck formatter extension

CTRL + SHIFT + p (opens the command pallet)
type settings.json (opens user settings)
set as a default formatter the black formater and max line length to 120

{
    "[python]": {
    "editor.defaultFormatter": "ms-python.black-formatter",
    },
    "black-formatter.args": ["--line-length", "120"]
}