DEV Community: Luiz Gustavo Erthal

Python Type Hints

Luiz Gustavo Erthal — Mon, 31 Mar 2025 22:51:48 +0000

Unlike C, Python is a dynamically typed programming language, which means you do not need to specify variable types in your code. The type of a variable is determined at runtime and can change while the program is running.

However, this flexibility can sometimes make code and logic less obvious. To address this, Python introduced Type Hints in version 3.5. Type hints allow us to specify the expected data type of variables, making the code more readable and understandable.

It’s important to note that I wrote "expected data type" rather than saying it's mandatory—because type hints are optional, as we’ll see later.

Example

def calc_triangle_area(base, height):
    return 0.5 * base * height

This simple function calculates the area of a triangle given a base and height. If we ask people to provide values, some might say 1, 2, 9, or 10, while others might say 3.5 or 10.5. All these answers are correct since triangles can have a wide range of base and height values.

However, we cannot accept values like strings (e.g., "3" and "8.5"), as they are not valid numerical inputs.

This is where Type Hints come in handy.

def calc_triangle_area(base: float, height: float) -> Float:
    return 0.5 * base * height

Even in a simple function like this, type hints improve readability by clearly indicating what types the variables (base and height) should have. Additionally, the -> float syntax specifies the expected return type.

One of the best things about Type Hints is that modern code editors (IDEs) will show expected types as you write the code. This makes debugging easier, improves logic validation, and helps ensure data quality across your project.

Creating a new Airbyte connector from scratch

Luiz Gustavo Erthal — Mon, 31 Mar 2025 22:05:46 +0000

In this tutorial, we will create a new Airbyte connector from scratch using the Airbyte UI Connector Builder.

You'll learn:

How to install Airbyte
How to create a new connector from a third-party API
How to set up data extraction
How to connect with BigQuery

1. Installing Airbyte locally

To install Airbyte locally on Ubuntu/MacOS, follow Airbyte's installation guide here.

You'll need Docker Desktop installed, then run these two commands:

Install abctl

curl -LsfS https://get.airbyte.com | bash -

Run Docker Desktop and then this line:

abctl local install

2. Weather API on Python

Before building our new connector, let's test the API and its parameters. For this tutorial, we'll use [WeatherAPI][https://www.weatherapi.com/]. You can create a free account and obtain an API key.

A sample of my Python script:

import requests

class WeatherAPI():

    def __init__(self, city, key, days):
        self.city = city,
        self.key = key,
        self.days = days

    def get_data(self):
        endpoint = 'http://api.weatherapi.com/v1/current.json'
        params = {
            'q': self.city,
            'days': self.days,
            'key': self.key
        }
        response = requests.get(endpoint, params=params)

        if response.status_code == 200:
            return response.json()
        else:
            return ValueError(f'Error fetching data from WeatherAPI. Error: {response.status_code}')

This WeatherAPI class takes a city, an API key, and the number of days for the weather forecast.

The expected JSON response:

{
    'location': 
        {
        'name': 'Florianopolis', 
        'region': 'Santa Catarina', 
        'country': 'Brazil', 
        'lat': -27.5833, 
        'lon': -48.5667, 
        'tz_id': 'America/Sao_Paulo', 
        'localtime_epoch': 1742939659, 
        'localtime': '2025-03-25 18:54'
        }, 
    'current': 
        {
        'last_updated_epoch': 1742939100, 
        'last_updated': '2025-03-25 18:45', 
        'temp_c': 26.4, 
        'temp_f': 79.5, 
        'is_day': 0, 
        'condition': 
            {
            'text': 'Partly cloudy', 
            'icon': '//cdn.weatherapi.com/weather/64x64/night/116.png', 
            'code': 1003
            }, 
            'wind_mph': 6.3, 
            'wind_kph': 10.1, 
            'wind_degree': 178, 
            'wind_dir': 'S', 
            'pressure_mb': 1015.0, 
            'pressure_in': 29.97, 
            'precip_mm': 0.75, 
            'precip_in': 0.03, 
            'humidity': 74, 
            'cloud': 75, 
            'feelslike_c': 30.4, 
            'feelslike_f': 86.7, 
            'windchill_c': 22.2, 
            'windchill_f': 72.0, 
            'heatindex_c': 24.7, 
            'heatindex_f': 76.4, 
            'dewpoint_c': 20.6, 
            'dewpoint_f': 69.1, 
            'vis_km': 10.0, 
            'vis_miles': 6.0, 
            'uv': 0.0, 
            'gust_mph': 10.2, 
            'gust_kph': 16.4
            }
        }
}

3. Connector Builder

Access Airbyte and navigate to Builder → Start from Scratch.

1. Start from scratch

2. Global Configuration

In API Base URL, paste the base URL of your API (excluding endpoint names).

For Authentication, choose API Key, select Query Parameter under Inject Into, and enter "key" as the parameter name. Different APIs have different authentication methods, so check the API documentation.

3. Stream

Create a new stream. In Airbyte, streams represent API endpoints. We'll use current.json and forecast.json from WeatherAPI.

In query parameters, add the required parameters. For this example, use q (location name) and set a default value, e.g., "Florianópolis".

4. Testing

Click Test to run the API request and check the response. If the response matches our expected JSON output, the configuration is correct.

Test different streams and parameters to validate everything.

Once ready, click Publish.

Creating a new connection

Now that we have our source (the connector), we need to connect it to a destination. We'll use the Google BigQuery connector.

Click Create a new connection (or Create your first connection if it's your first time), then:

Select a source (our WeatherAPI connector)
Select a destination (Google BigQuery)

Click in "Custom" and select your source connector.

You can follow the guide from how to setup your Google BigQuery that Airbyte show to you.

Project ID: Your Google Cloud project name
Dataset Location: Region where you store data (e.g., us-east-1)
Default Dataset ID: it's the dataset name
Loading Method: choose the one you prefer, I'll be selecting batched
Service Account Key: Paste the JSON value of your service account

Select the streams you need to extract data. For sync mode, select the best that suits you. For our example, as I'm getting the current weather for each day that I request the data, I'll be doing a full refresh and appending data.

In this phase, you can give a better name for your connection. I'll be leaving the default option.

For schedule type, it's the time that the connection will run. As it's a current weather for daily forecast, I'll be leaving it as every 24 hours.

The destination namespace will change for the name conventions your company use. For example, if your company use the Raw -> Source -> Staging -> Mart convention, this would be something like raw_weather.

Click in finish and sync.

5. Sync your data

Airbyte will now run the first sync and load data into BigQuery (or another configured destination). This process may take time, depending on your dataset size.

Once completed, you can query your BigQuery dataset to verify the imported data.

Testing Google Cloud Functions Locally

Luiz Gustavo Erthal — Mon, 17 Mar 2025 23:31:43 +0000

Google Cloud Functions is a serverless compute platform that allows you to run code in response to events without provisioning or managing servers.

In this first part of a two-step tutorial, we’ll explore how to test Google Cloud Functions locally. In the second part, we’ll cover how to deploy them from GitHub.

Prerequisites

To follow along, you'll need:

A Google Cloud account
A Service Account with permissions to deploy Google Cloud Functions
Some code you want to run on Cloud Functions

For this example, I'll use the following sample code:

import requests
import pandas_gbq
import pandas as pd
from google.oauth2 import service_account 


def get_currency():
    """
    Get currency exchange rates from the API and call the save_to_bigquery function to save the data to BigQuery.
    Returns the data as a DataFrame.
    """
    url = "https://open.er-api.com/v6/latest/USD" 
    response = requests.get(url=url)

    data = response.json()
    df = pd.DataFrame(list(data['rates'].items()), columns=["currency", "exchange_rate"])

    if not df.empty:
        save_to_bigquery(df)
        return "Process completed!"
    else:
        return "No data found."


def save_to_bigquery(dataframe):
    """
    Save the data to BigQuery.
    Required: service_account.json file with the credentials.
    """
    df = dataframe
    project_id = "erthal-blog-de-projects" 
    dataset_table = "financial.currency"
    credentials = service_account.Credentials.from_service_account_file("service_account.json")

    pandas_gbq.to_gbq(df, dataset_table, project_id, credentials = credentials, if_exists='replace')

You can use any code you like—this is just a simple example. The script above fetches currency exchange rates (USD to other currencies) from a free API and stores the data in a BigQuery dataset.

Functions Framework

To test the function locally, we can use Google's function framework. This library allows us to set up a local development server and trigger the function via HTTP requests.

Install the lib via pip:

pip install functions-framework

Setting up the HTTP Server

To run the function locally, we need to:

Import the library
Define an entry point for the function

Importing the library

import functions-framework

Defining the entry point

The @functions_framework.http decorator specifies where the function should start handling requests.

@functions_framework.http
def get_currency(request):
    """
    Get currency exchange rates from the API and call the save_to_bigquery function to save the data to BigQuery.
    Returns the data as a DataFrame.
    """
    url = "https://open.er-api.com/v6/latest/USD" 
    response = requests.get(url=url)

    data = response.json()
    df = pd.DataFrame(list(data['rates'].items()), columns=["currency", "exchange_rate"])

    if not df.empty:
        save_to_bigquery(df)
        return "Process completed!"
    else:
        return "No data found."

Note: I added the request parameter in get_currency(request), as Google Cloud Functions require it to process HTTP requests.

Running the Function Locally

Start the local server by running:

functions-framework --target=main --port=YOUR_PORT

This command starts an HTTP server on the specified port (default: 8080). If the command runs without errors, the server is up and waiting for requests.

Stopping the Function

If you need to stop the server (e.g., to redeploy your function), use the following command on Ubuntu:

fuser -k YOUR_PORT/tcp

Testing the Function

To send a request to your local function, open another terminal (or use Postman) and run:

curl localhost:YOUR_PORT -X POST -H "Content-Type: application/json"

If everything is set up correctly, the function should return a successful response (HTTP 200) and process the request. You can monitor the logs in the terminal where the function is running.

I hope this guide helps you test Google Cloud Functions locally! 🚀

How to Create Python Virtual Environments on Ubuntu

Luiz Gustavo Erthal — Fri, 21 Feb 2025 22:34:39 +0000

When working on different Python projects, it's often necessary to create isolated environments for specific tasks. This is where Python's Virtual Environments package comes in handy.

Use this post as a guide or a cheat sheet for future reference.

Creating a Virtual Environment

To create a new virtual environment, use the following command:

python3 -m venv .venv

The .venv directory will contain your virtual environment. Using a dot (.) before the name makes it a hidden folder, which is a common practice. However, you can name it anything you like to suit your project structure.

Activating the Virtual Environment

To activate the virtual environment, run:

source .venv/bin/activate

Once activated, you can install any required packages and libraries. These installations will be isolated within the .venv directory and will not affect global Python packages.

For example, to install pandas:

pip install pandas

Deactivating the Virtual Environment

To deactivate the virtual environment, simply run:

deactivate

Listing Installed Packages

While the virtual environment is activated, you can list all installed packages using:

List

pip list

pip freeze

The only difference between each command is that list will display a human-readable list while freeze will outputs installed packages in a machine-readable format.

Sharing Your Virtual Environment

If you need to share your project, ensure that others can replicate your environment. You can do this by exporting the installed packages to a requirements.txt file:

Save your venv into a .txt

pip freeze > requirements.txt

If you receive a requirements.txt file from a colleague or friend, you can install all required packages by running:

Import the packages from a requirements.txt

pip install -r requirements.txt

By following these steps, you can effectively manage your Python environments and ensure consistency across different projects.