SQL Server e Python com método CRUD

O objetivo deste post é manipular um banco de dados SQLServer com Python.

Criaremos um ambiente virtual usando o VSCode e este ambiente será o interpretador do Python.

🔵 Para instalar o VSCode, clique aqui

🔵 Para instalar a linguagem Python no seu computador, clique aqui

🔵 Para criar um ambiente virtual e um arquivo requirements.txt veja este post clique aqui

Após preparar o ambiente, instale a biblioteca no terminal (novo terminal) VSCode ou no editor que estiver utilizando. Obs: Não é necessário criar um ambiente virtual, não é necessário utilizar o VSCode, você poderá utilizar o IDE que desejar.

 ◼ Instalar conector pyodbc

pip install pyodbc

Após instalar o conector:

◼ instale a linguagem SQL no seu computador clicando aqui

◼ será necessário também instalar o SGBD Management Studio clicando aqui

Obs: Instale a última versão. No momento deste post, a última versão era 19.3

Neste post mostro mais detalhes da instalação

No meu caso, instalei o SQL Express, criei um banco de dados chamado empresa:

Após criar uma database:

Clique em Empresa, e posteriormente clique em "Nova consulta", e execute o comando abaixo pressionando o F5

-- Criando Tabela pessoa

CREATE TABLE pessoa (

id INT PRIMARY KEY IDENTITY,

nome VARCHAR(255) NOT NULL,

cpf VARCHAR(11) NOT NULL UNIQUE,

data_nascimento DATE NOT NULL

);

-- Criando Tabela conta

CREATE TABLE conta (

id INT PRIMARY KEY IDENTITY,

pessoa_id INT NOT NULL,

saldo NUMERIC(10, 2) NOT NULL DEFAULT 0,

ativo BIT NOT NULL DEFAULT 1,

FOREIGN KEY (pessoa_id) REFERENCES pessoa(id)

);

Com a nossa database criada e as tabelas criadas e definidas, voltamos ao Vscode no arquivo db.py

para

◼ testar a conexão:

# Parâmetros de conexão

server = 'localhost'

database = 'NomeDoBancoDeDados'

username = 'seuUsuario'

password = 'suaSenha'

No meu caso, como o login esta sendo autenticado pelo windows e não possui senha de usuário para conectar no sgbd, só utilizei o nome do servidor (este nome é mostrado no ato da instalação do sql)

🎯 Testar conexão

import pyodbc

# Parâmetros de conexão
server = 'AMANDA\SQLEXPRESS01'
database = 'EMPRESA'

# String de conexão
conn_str = f'DRIVER={{SQL Server}};SERVER={server};DATABASE={database};Trusted_Connection=yes;'

try:
    # Estabelecer conexão
    conexao = pyodbc.connect(conn_str)

    print('Conexão bem-sucedida!')

    # Executar consultas ou operações de banco de dados aqui...

except Exception as e:
    print('Erro ao conectar ao SQL Server:')
    print(e)

finally:
    # Certificar-se de fechar a conexão quando terminar
    if 'conexao' in locals():
        conexao.close()
        print('Conexão encerrada.')

🎯 Inserir dados

import pyodbc

# Parâmetros de conexão
server = 'AMANDA\SQLEXPRESS01'
database = 'EMPRESA'

# String de conexão
conn_str = f'DRIVER={{SQL Server}};SERVER={server};DATABASE={database};Trusted_Connection=yes;'

try:
    # Estabelecer conexão
    conexao = pyodbc.connect(conn_str)
    cursor = conexao.cursor()

    # Dados para inserção
    dados_insercao = [
        ('Fulano', '12345678900', '1990-01-01'),
        ('Ciclano', '98765432100', '1985-05-15'),
        ('Beltrano', '45678912300', '2000-10-20')
    ]

    # Query SQL para inserir dados na tabela pessoa
    query_insercao = "INSERT INTO pessoa (nome, cpf, data_nascimento) VALUES (?, ?, ?)"

    # Executar a inserção para cada conjunto de dados
    for dado in dados_insercao:
        cursor.execute(query_insercao, dado)

    # Confirmar a operação
    conexao.commit()

    print('Dados inseridos com sucesso!')
except Exception as e:
    print('Erro ao inserir dados:')
    print(e)

finally:
    # Certificar-se de fechar a conexão quando terminar
    if 'conexao' in locals():
        conexao.close()
        print('Conexão encerrada.')

Inserindo dados utilizando o input (dado que o próprio usuário insere):

import pyodbc

# Parâmetros de conexão
server = 'AMANDA\SQLEXPRESS01'
database = 'EMPRESA'

# String de conexão
conn_str = f'DRIVER={{SQL Server}};SERVER={server};DATABASE={database};Trusted_Connection=yes;'

try:
    # Estabelecer conexão
    conexao = pyodbc.connect(conn_str)
    cursor = conexao.cursor()

    # Solicitar ao usuário que insira os dados
    nome = input("Digite o nome: ")
    cpf = input("Digite o CPF: ")
    data_nascimento = input("Digite a data de nascimento (no formato AAAA-MM-DD): ")

    # Query SQL para inserir dados na tabela pessoa
    query_insercao = "INSERT INTO pessoa (nome, cpf, data_nascimento) VALUES (?, ?, ?)"

    # Executar a inserção
    cursor.execute(query_insercao, (nome, cpf, data_nascimento))

    # Confirmar a operação
    conexao.commit()

    print('Dados inseridos com sucesso!')
except Exception as e:
    print('Erro ao inserir dados:')
    print(e)

finally:
    # Certificar-se de fechar a conexão quando terminar
    if 'conexao' in locals():
        conexao.close()
        print('Conexão encerrada.')

🎯 Excluir dados através de um ID

import pyodbc

# Parâmetros de conexão
server = 'AMANDA\SQLEXPRESS01'
database = 'EMPRESA'

# String de conexão
conn_str = f'DRIVER={{SQL Server}};SERVER={server};DATABASE={database};Trusted_Connection=yes;'

try:
    # Estabelecer conexão
    conexao = pyodbc.connect(conn_str)
    cursor = conexao.cursor()

    # Solicitar ao usuário que insira o ID para exclusão
    id_excluir = input("Digite o ID que deseja excluir: ")

    # Query SQL para excluir todos os dados do ID especificado
    query_exclusao = "DELETE FROM pessoa WHERE id = ?"

    # Executar a exclusão
    cursor.execute(query_exclusao, (id_excluir,))

    # Confirmar a operação
    conexao.commit()

    print(f'Todos os dados do ID {id_excluir} foram excluídos com sucesso!')
except Exception as e:
    print('Erro ao excluir dados:')
    print(e)

finally:
    # Certificar-se de fechar a conexão quando terminar
    if 'conexao' in locals():
        conexao.close()
        print('Conexão encerrada.')

🎯 Alterar o cpf de um ID através de input

import pyodbc

# Parâmetros de conexão
server = 'AMANDA\SQLEXPRESS01'
database = 'EMPRESA'

# String de conexão
conn_str = f'DRIVER={{SQL Server}};SERVER={server};DATABASE={database};Trusted_Connection=yes;'

try:
    # Estabelecer conexão
    conexao = pyodbc.connect(conn_str)
    cursor = conexao.cursor()

    # Solicitar ao usuário que insira o ID para atualização
    id_atualizar = input("Digite o ID que deseja atualizar: ")
    novo_cpf = input("Digite o novo CPF: ")

    # Query SQL para atualizar o CPF do ID especificado
    query_atualizar_cpf = "UPDATE pessoa SET cpf = ? WHERE id = ?"

    # Executar a atualização
    cursor.execute(query_atualizar_cpf, (novo_cpf, id_atualizar))

    # Confirmar a operação
    conexao.commit()

    print(f'CPF do ID {id_atualizar} atualizado com sucesso!')
except Exception as e:
    print('Erro ao atualizar CPF:')
    print(e)

finally:
    # Certificar-se de fechar a conexão quando terminar
    if 'conexao' in locals():
        conexao.close()
        print('Conexão encerrada.')

Antes de alterar o ID, perguntar ao usuário se realmente deseja alterar o cpf de {nome}

import pyodbc

# Parâmetros de conexão
server = 'AMANDA\SQLEXPRESS01'
database = 'EMPRESA'

# String de conexão
conn_str = f'DRIVER={{SQL Server}};SERVER={server};DATABASE={database};Trusted_Connection=yes;'

try:
    # Estabelecer conexão
    conexao = pyodbc.connect(conn_str)
    cursor = conexao.cursor()

    # Solicitar ao usuário que insira o ID para atualização
    id_atualizar = input("Digite o ID que deseja atualizar: ")

    # Consulta para obter o nome associado ao ID fornecido pelo usuário
    query_consulta_nome = "SELECT nome FROM pessoa WHERE id = ?"
    cursor.execute(query_consulta_nome, (id_atualizar,))
    resultado = cursor.fetchone()

    if resultado:
        nome = resultado[0]
        confirmacao = input(f"Tem certeza que deseja alterar o ID {id_atualizar} de nome {nome}? (s/n): ")
        
        if confirmacao.lower() == 's':
            novo_cpf = input("Digite o novo CPF: ")

            # Query SQL para atualizar o CPF do ID especificado
            query_atualizar_cpf = "UPDATE pessoa SET cpf = ? WHERE id = ?"

            # Executar a atualização
            cursor.execute(query_atualizar_cpf, (novo_cpf, id_atualizar))

            # Confirmar a operação
            conexao.commit()

            print(f'CPF do ID {id_atualizar} atualizado com sucesso!')
        else:
            print('Operação cancelada pelo usuário.')
    else:
        print(f'ID {id_atualizar} não encontrado na tabela.')

except Exception as e:
    print('Erro ao atualizar CPF:')
    print(e)

finally:
    # Certificar-se de fechar a conexão quando terminar
    if 'conexao' in locals():
        conexao.close()
        print('Conexão encerrada.')

Antes de alterar,perguntar ao usuário o que ele deseja alterar e se realmente deseja alterar o {nome}. Acrescentei o while para que o programa continue rodando:

import pyodbc

# Parâmetros de conexão
server = 'AMANDA\SQLEXPRESS01'
database = 'EMPRESA'

# String de conexão
conn_str = f'DRIVER={{SQL Server}};SERVER={server};DATABASE={database};Trusted_Connection=yes;'

try:
    while True:
        # Estabelecer conexão
        conexao = pyodbc.connect(conn_str)
        cursor = conexao.cursor()

        # Opções de alteração disponíveis
        opcoes_alteracao = ['nome', 'cpf', 'data de nascimento']

        # Solicitar ao usuário que escolha o que deseja alterar
        print("Opções de alteração disponíveis:")
        for i, opcao in enumerate(opcoes_alteracao, start=1):
            print(f"{i}. {opcao.capitalize()}")

        escolha = int(input("Escolha o número correspondente ao que deseja alterar: "))

        # Mapear a escolha do usuário para o nome da coluna na tabela
        coluna = opcoes_alteracao[escolha - 1]

        # Solicitar ao usuário que insira o ID para atualização
        id_atualizar = input("Digite o ID que deseja atualizar: ")

        # Consulta para obter o nome associado ao ID fornecido pelo usuário
        query_consulta_nome = "SELECT nome FROM pessoa WHERE id = ?"
        cursor.execute(query_consulta_nome, (id_atualizar,))
        resultado = cursor.fetchone()

        if resultado:
            nome = resultado[0]
            confirmacao = input(f"Tem certeza que deseja alterar o ID {id_atualizar} de nome {nome}? (s/n): ")
            
            if confirmacao.lower() == 's':
                novo_valor = input(f"Digite o novo valor para {coluna}: ")

                # Query SQL para atualizar o valor da coluna especificada
                query_atualizar_valor = f"UPDATE pessoa SET {coluna} = ? WHERE id = ?"

                # Executar a atualização
                cursor.execute(query_atualizar_valor, (novo_valor, id_atualizar))

                # Confirmar a operação
                conexao.commit()

                print(f'{coluna.capitalize()} do ID {id_atualizar} atualizado com sucesso!')
            else:
                print('Operação cancelada pelo usuário.')
        else:
            print(f'ID {id_atualizar} não encontrado na tabela.')

        # Perguntar ao usuário se deseja continuar
        continuar = input("Deseja continuar? [s/n]: ")
        if continuar.lower() != 's':
            break

except Exception as e:
    print('Erro ao atualizar valor:')
    print(e)

finally:
    # Certificar-se de fechar a conexão quando terminar
    if 'conexao' in locals():
        conexao.close()
        print('Conexão encerrada.')