Como acessar uma API de endereços com Python e gravar em um banco de dados SQL Server

Neste artigo, vamos aprender uma forma de acessar uma API de endereços com Python e gravar os dados em um banco de dados SQL Server.

Com o código abaixo , vamos puder acessar a API e pegar todas as informações de endereço.

Como entrada, tenho uma lista com CEP(s). Essa lista, poderia ser um arquivo ou um campo de um site de vendas, por exemplo.

Esse é o site onde vamos acessar os endereços: Via cep

Os dados obtidos, serão armazenados em um SQL Server.

O SQL Server é um sistema de gerenciamento de banco de dados relacional desenvolvido pela Microsoft. Ele é usado para armazenar e recuperar dados conforme necessário por outras aplicações.

Os dados de conexão do SQL Server, você pode encontrar na tela inicial do SQL Server.

As bibliotecas utilizadas para este projeto são:

O pandas com objetivo de criar uma lista e transformar a lista em um DataFrame;
request: tem como finalidade tornar as solicitações HTTP mais amigáveis e simples.
pyodbc: é uma biblioteca Python de código aberto que fornece conectividade ao SQL Server. Ele usa o padrão Open Database Connectivity (ODBC) para interagir com o banco de dados, tornando-o uma ferramenta versátil que pode conectar o Python a vários sistemas de banco de dados.

Segue o script completo:

import requests
import pandas as pd
import pyodbc

# Recebendo como entrada uma lista de CEP(s)
lista_de_ceps: list = ['17021600', '17013221', '17015011', '17015021']

# Criando uma lista vazia para armazenar os endereços
lista_de_enderecos: list = []

for cep in lista_de_ceps:

    url: str = 'https://viacep.com.br/ws/{}/json/'.format(cep)

    try:
        req = requests.get(url, timeout=3)
        if req.status_code == 200:
            endereco = req.json()
            lista_de_enderecos.append(
                [
                    endereco['cep'],
                    endereco['logradouro'],
                    endereco['complemento'],
                    endereco['bairro'],
                    endereco['localidade'],
                    endereco['uf']
                ]
            )

        else:
            erro = req.raise_for_status()
            print(f'Erro no acesso a API: {erro}')

    except Exception as erro:
        print(f'Erro na execução do código: {erro}')
print('Lista de endereços capturada')

# Mostrando os dados do DataFrame
for item in lista_de_enderecos:
    print(item)

# Transformando a lista de endereços em um DataFrame

df_enderecos = pd.DataFrame(lista_de_enderecos,
                            columns=['cep','logradouro','complemento','bairro','localidade','uf'])

# Preparando a conexão e gravando em uma tabela no banco de dados.
# Substitua o server com o nome do seu servidor e o Database com 
# o nome do seu banco de dados
dados_conexao = (
    "Driver={SQL Server};""Server=NOTEBOOK-SAMSUN;""Database=Zungando_Microsoft;"
)

conexao = pyodbc.connect(dados_conexao)
print('Database connection ok!')

# Deleta a tabela existente, se houver
conexao.execute('DROP TABLE IF EXISTS Enderecos')

# Criar uma tabela para armazenar os dados de endereço

conexao.execute(''' CREATE TABLE Enderecos(
cep VARCHAR(20),
logradouro VARCHAR(30),
complemento VARCHAR(100),
bairro VARCHAR(50),
localidade VARCHAR(50),
uf VARCHAR(50)                             
)''')


# Grava e fecha a conexão
conexao.commit()
conexao.close()

# conexão aberta, consome recursos do banco de dados
# Conecta ao banco de dados
conexao = pyodbc.connect(dados_conexao)
print('Insert Into Enderecos - Database connection OK!')

# Insere cada linha do arquivo na tabela do banco de dados
for index, row in df_enderecos.iterrows():
    conexao.execute(
        'INSERT INTO Enderecos(cep,logradouro,complemento,bairro,localidade,uf)VALUES(?,?,?,?,?,?)',
        row.cep, row.logradouro, row.complemento, row.bairro, row.localidade, row.uf)
conexao.commit()
conexao.close()

print("Carga Realizada!")