Python Code Examples for Connecting to Common Databases

Connecting to various databases is a frequent task in Python development; the following examples demonstrate how to establish connections, execute simple SELECT queries, retrieve results, and properly close each connection using the appropriate library for each database system.

1. MySQL

import mysql.connector
# Connect to database
cnx = mysql.connector.connect(user='username', password='password',
                              host='localhost', database='database_name')
# Execute query
cursor = cnx.cursor()
query = "SELECT * FROM table_name"
cursor.execute(query)
# Get results
for row in cursor:
    print(row)
# Close connection
cursor.close()
cnx.close()

2. PostgreSQL

import psycopg2
# Connect to database
conn = psycopg2.connect(database="database_name", user="username", password="password", host="localhost", port="5432")
# Execute query
cur = conn.cursor()
cur.execute("SELECT * FROM table_name")
rows = cur.fetchall()
for row in rows:
    print(row)
# Close connection
cur.close()
conn.close()

3. SQLite

import sqlite3
# Connect to database
conn = sqlite3.connect('database_name.db')
# Execute query
cur = conn.cursor()
cur.execute("SELECT * FROM table_name")
rows = cur.fetchall()
for row in rows:
    print(row)
# Close connection
cur.close()
conn.close()

4. MongoDB

from pymongo import MongoClient
# Connect to database
client = MongoClient('mongodb://localhost:27017/')
# Get database and collection
db = client['database_name']
collection = db['collection_name']
# Query data
for document in collection.find():
    print(document)
# Close connection
client.close()

5. Oracle

import cx_Oracle
# Connect to database
conn = cx_Oracle.connect('username/password@host:port/service_name')
# Execute query
cur = conn.cursor()
cur.execute("SELECT * FROM table_name")
rows = cur.fetchall()
for row in rows:
    print(row)
# Close connection
cur.close()
conn.close()

6. Microsoft SQL Server

import pyodbc
# Connect to database
conn = pyodbc.connect('DRIVER={SQL Server};SERVER=server_name;DATABASE=database_name;UID=username;PWD=password')
# Execute query
cur = conn.cursor()
cur.execute("SELECT * FROM table_name")
rows = cur.fetchall()
for row in rows:
    print(row)
# Close connection
cur.close()
conn.close()

7. Redis

import redis
# Connect to database
r = redis.Redis(host='localhost', port=6379, db=0)
# Store data
r.set('key', 'value')
# Retrieve data
value = r.get('key')
print(value)
# Close connection (optional)
r.close()

8. Elasticsearch

from elasticsearch import Elasticsearch
# Connect to database
es = Elasticsearch(['localhost:9200'])
# Query data
res = es.search(index='index_name', body={'query': {'match_all': {}}})
for hit in res['hits']['hits']:
    print(hit['_source'])
# Close connection (optional)
es.close()

9. Cassandra

from cassandra.cluster import Cluster
# Connect to database
cluster = Cluster(['localhost'])
session = cluster.connect('keyspace_name')
# Execute query
rows = session.execute("SELECT * FROM table_name")
for row in rows:
    print(row)
# Close connection
session.shutdown()
cluster.shutdown()

10. Couchbase

from couchbase.cluster import Cluster
from couchbase.cluster import PasswordAuthenticator
# Connect to database
cluster = Cluster('couchbase://localhost')
authenticator = PasswordAuthenticator('username', 'password')
cluster.authenticate(authenticator)
bucket = cluster.open_bucket('bucket_name')
# Execute query
query = bucket.query("SELECT * FROM bucket_name")
for row in query:
    print(row)
# Close connection
cluster.disconnect()

11. Firebase Realtime Database

import firebase_admin
from firebase_admin import credentials, db
# Initialize Firebase app
cred = credentials.Certificate('path/to/serviceAccountKey.json')
firebase_admin.initialize_app(cred, {
    'databaseURL': 'https://your-database-url.firebaseio.com'
})
# Get database reference
ref = db.reference('path/to/data')
# Read data
data = ref.get()
print(data)
# Close connection (optional)
firebase_admin.delete_app(firebase_admin.get_app())

12. Amazon DynamoDB

import boto3
# Connect to database
dynamodb = boto3.resource('dynamodb', region_name='us-west-2')
# Get table
table = dynamodb.Table('table_name')
# Query data
response = table.get_item(Key={'key_name': 'key_value'})
item = response['Item']
print(item)
# Close connection (optional)
dynamodb.close()

13. Neo4j Graph Database

from neo4j import GraphDatabase
# Connect to database
driver = GraphDatabase.driver("bolt://localhost:7687", auth=("username", "password"))
with driver.session() as session:
    result = session.run("MATCH (n) RETURN n LIMIT 10")
    for record in result:
        print(record)
# Close connection
driver.close()

14. InfluxDB Time‑Series Database

from influxdb import InfluxDBClient
# Connect to database
client = InfluxDBClient(host='localhost', port=8086, username='username', password='password', database='database_name')
# Query data
result = client.query('SELECT * FROM measurement_name LIMIT 10')
for point in result.get_points():
    print(point)
# Close connection
client.close()

These examples illustrate how to use the appropriate Python libraries to connect to, query, and close connections for a wide range of relational, NoSQL, and time‑series databases.