Da immer wieder die Frage auftaucht, wie man von Python aus auf PostgreSQL zugreifen kann, habe ich mich dazu entschlossen, ein allgemeines Beispiel dafür zu erstellen.
Die Erklärungen sind im Quellcode als Kommentare enthalten. Bitte Quellcode zwei mal durchlesen und drei mal ausprobieren, bevor hier eine Frage gestellt wird, die sowiso bereits im Quellcode beantwortet wurde.
Code: Alles auswählen
#!/usr/bin/env python
# coding: utf-8
import psycopg2
import psycopg2.extensions
import datetime
# Mit Unicode arbeiten.
#
# ERKLÄRUNG!
#
# PostgreSQL arbeitet normalerweise mit UTF-8 als Encoding. Übergeben wir
# einen Unicode-String, dann wird dieser automatisch nach UTF-8 umgewandelt.
# Übergeben wir einen Bytestring, dann nimmt psycopg2 an, das es sich um einen
# String im Encoding UTF-8 handelt. --> Um Problemen mit Umlauten aus dem Weg zu
# gehen, sollten wir uns immer darum kümmern, dass keine Bytestrings, sondern
# Unicode übergeben wird. Immer!
#
# Dieser Fehler:
#
# ProgrammingError: invalid byte sequence for encoding "UTF8": 0xXX
# HINT: This error can also happen if the byte sequence does not match
# the encoding expected by the server, which is controlled by "client_encoding".
#
# taucht meistens dann auf, wenn wir uns nicht korrekt darum gekümmert haben,
# dass nur Unicode übergeben wird.
#
# Wenn wir Textdaten aus der Datenbank auslesen, bekommen wir einen
# in UTF-8 codierten String zurück. Damit wir uns nicht selber um die Decodierung
# kümmern müssen, also automatisch Unicode zurück bekommen, müssen wir
# die UNICODE-Erweiterung registrieren.
psycopg2.extensions.register_type(psycopg2.extensions.UNICODE)
def drop_table(conn):
"""
Löscht die Adressentabelle, falls diese existiert.
Es handelt sich ja nur um ein Beispiel. Und damit dieses Beispiel öfter
ausgeführt werden kann, lösche ich die Tabelle.
"""
# Cursor erstellen
cur = conn.cursor()
# SQL-Anweisung zum Löschen einer Tabelle
sql = """
DROP TABLE test_addresses
"""
try:
# SQL-Anweisung ausführen und danach mit ``commit`` bestätigen.
#
# Erklärung: SQL-Anweisungen werden in einer Transaktion zusammengefasst.
# Die Anweisungen werden erst dann gültig, wenn die gesamte
# Transaktion mit ``commit`` bestätigt wurde. Mit ``commit` wird automatisch
# auch eine neue, frische Transaktion erstellt, die für die nächsten
# SQL-Anweisungen benutzt werden kann.
cur.execute(sql)
conn.commit()
print "table created"
except psycopg2.ProgrammingError, errval:
# Falls die Tabelle bereits existiert, tritt der Fehler
# ``psycopg2.ProgrammingError`` auf.
# Da in diesem Fall die SQL-Anweisung nicht ausgeführt werden kann, muss
# die angefangene Transaktion rückgängig gemacht werden. Das geschieht
# mit ``rollback()``. Damit wird automatisch auch wieder eine neue,
# frische Transaktion erstellt, die für die nächsten SQL-Anweisungen
# benutzt werden kann.
conn.rollback()
print errval
def create_table(conn):
"""
Erstellt eine Tabelle für Adressen. Indexe werden erstellt, damit die
Tabelle schneller durchsucht werden kann.
"""
# Cursor erstellen
cur = conn.cursor()
# SQL-Anweisung zum Erstellen einer Tabelle mit automatisch hochzählendem
# Primärschlüssel. Dieser Primärschlüssel identifiziert einen
# Datensatz (=Datenzeile) eindeutig. PostgreSQL macht bei der Länge von
# Textfeldern keinen Unterschied. Es ist egal, ob nur eine kurze PLZ oder
# ein langer Text (bis 1GB) darin liegt.
# ``NOT NULL`` gibt an, dass in diesem Feld unbedingt etwas stehen muss.
# ``WITHOUT OIDS`` gibt an, dass PostgreSQL sich nicht darum kümmern soll,
# eine datenbankweit eindeutige ID für die Datensätze zu erstellen.
sql = """
-- Tabelle erstellen
CREATE TABLE test_addresses (
id SERIAL PRIMARY KEY,
name TEXT NOT NULL,
name2 TEXT,
first_name TEXT,
street_nr TEXT,
zip_code TEXT,
town TEXT,
birthday DATE,
notice TEXT
) WITHOUT OIDS;
-- Indizes erstellen
CREATE INDEX ix_test_addresses_name ON test_addresses (name);
CREATE INDEX ix_test_addresses_first_name ON test_addresses (first_name);
CREATE INDEX ix_test_addresses_zip_code ON test_addresses (zip_code);
CREATE INDEX ix_test_addresses_town ON test_addresses (town);
"""
# SQL-Anweisungen ausführen
cur.execute(sql)
# Transaktion bestätigen.
# Mit ``commit` wird automatisch auch eine neue, frische Transaktion
# erstellt, die für die nächsten SQL-Anweisungen benutzt werden kann.
conn.commit()
def insert_address(
conn, name, name2 = None, first_name = None, street_nr = None, zip_code = None,
town = None, birthday = None, notice = None
):
"""
Fügt der Tabelle 'test_adresses' einen neuen Datensatz hinzu
"""
# Cursor erstellen
cur = conn.cursor()
# SQL-Anweisung erstelltn
#
# HINWEIS!!!
#
# Die Platzhalter (%s) werden von **psycopg2** automatisch befüllt.
# Man darf die Platzhalter NICHT in Anführungszeichen (') einschließen.
# Diese werden von psycopg2 automatisch eingefügt,
# wenn es sich um einen String handelt.
sql = """
INSERT INTO test_addresses (
name, name2, first_name, street_nr, zip_code, town, birthday, notice
) VALUES (
%s, %s, %s, %s, %s, %s, %s, %s
)
"""
# SQL-Anweisung ausführen.
#
# HINWEISE!!!
#
# Dabei werden in einem Iterable (z.B. Liste oder Tuppel)
# die Werte in der richtigen Reihenfolge übergeben. *psycopg2* kümmert sich
# darum, dass die Daten korrekt eingefügt werden. Z.B. wird bei None ein NULL
# übergeben. Wenn es ein String ist, wird dieser korrekt in Anführungszeichen
# eingegrenzt. Wenn es UNICODE ist, dann wird der Text automatisch in das
# Encoding der Datenbank umgewandelt. Wenn es ein datetime-Objekt ist, dann
# wird ein gültiges SQL-Datum daraus. Wenn es ein psycopg2.Binary-Objekt ist,
# dann wird damit ein Datenstream übergeben. Usw.
cur.execute(sql, (name, name2, first_name, street_nr, zip_code, town, birthday, notice))
# Transaktion bestätigen
conn.commit()
print "address inserted"
def insert_addresses(conn, new_addresses):
"""
Fügt alle Adressen in einem Rutsch hinzu.
"""
# Diesmal ohne viele Erklärungen...
cur = conn.cursor()
sql = """
INSERT INTO test_addresses (
name, name2, first_name, street_nr, zip_code, town, birthday, notice
) VALUES (
%s, %s, %s, %s, %s, %s, %s, %s
)
"""
cur.executemany(sql, new_addresses)
conn.commit()
print "addresses inserted"
def get_all_addresses(conn):
"""
Gibt alle Adressen in/mit einem Cursor zurück.
"""
# Ohne viele Erklärungen...
cur = conn.cursor()
sql = """
SELECT
id, name, name2, first_name, street_nr, zip_code, town, birthday
FROM
test_addresses
ORDER BY
name, name2, first_name
"""
cur.execute(sql)
return cur
def get_one_address(conn, id):
"""
Gibt die Adresse mit der angegebenen ID zurück.
Ja, ich weiß, dass ich die eingebaute Funktion ``id()`` überschreibe.
Und da ich id() noch nie in einem Programm verwendet habe und noch keinen
Ersatz für diesen Namen gefunden habe, ist es mir egal. :-)
"""
cur = conn.cursor()
sql = """
SELECT
id, name, name2, first_name, street_nr, zip_code, town, birthday, notice
FROM
test_addresses
WHERE
(id = %s)
"""
# SQL-Anweisung ausführen
#
# HINWEIS!!!
#
# Auch wenn nur ein einziger Wert übergeben wird. Es muss ein Tuppel, eine
# Liste oder ein anderes Iterable mit dem Wert übergeben werden. Und nicht
# nur der Wert alleine.
cur.execute(sql, (id, ))
# Nur eine Zeile als Tuppel zurück geben
return cur.fetchone()
def main():
# Connection zur Datenbank öffnen
conn = psycopg2.connect(
database = "testdb",
host = "localhost",
user = "tester",
password = "tester"
)
# Tabelle löschen falls sie existiert
drop_table(conn)
# Tabelle erstellen
create_table(conn)
# Eine Adresse hinzufügen
insert_address(
conn, name =u"Mustermann", first_name = u"Thomas",
street_nr = u"Überbachweg 15", birthday = datetime.date(1980, 10, 22),
notice = u"Das ist der Typ der ständig rum motzt."
)
# Mehrere Adressen in einem Rutsch hinzufügen
new_addresses = (
(
u"Waller", None, u"Heinz", u"Lampenweg 5", "6020", "Innsbruck",
datetime.date(1974, 8, 18), None
),
(
u"Haller", None, u"Karl", None, None, None, datetime.date(1974, 8, 18), None
),
(
u"Mistkübel AG", u"Annahmestelle", None, u"Fönweg 15", u"6020",
u"Innsbruck", None, None
),
(
u"Ordination Dr. Mustermann", u"Ärztehaus", None, u"Zipperlein 3", u"6410",
u"Telfs", None, None
),
)
insert_addresses(conn, new_addresses)
# Alle Adressen anzeigen
cur = get_all_addresses(conn)
for (id, name, name2, first_name, street_nr, zip_code, town, birthday) in cur:
print
print "ID:", id
print "Name:", repr(name)
print "Name2:", repr(name2)
print "Vorname:", repr(first_name)
print "Straße Nr.:", repr(street_nr)
print "PLZ:", repr(zip_code)
print "Ort:", repr(town)
print "Geburtstag:", repr(birthday)
# Eine bestimmte Adresse anzeigen
address = get_one_address(conn, 1)
print
print repr(address)
# Connection schließen
conn.close()
if __name__ == "__main__":
main()http://www.python-forum.de/topic-5095.html
Und etwas allgemeiner gehalten ist dieses Codebeispiel: http://www.python-forum.de/topic-11615.html
Stichworte: Datenbank Unicode Encoding UTF-8 PostgreSQL psycopg psycopg2
Bitte erstellt einen neuen Topic, falls Fragen zu diesem Beispiel auftauchen. Ich möchte dieses Beispiel nicht auslagern. Dann könnte man hier im Forum nicht mehr nach darin enthaltenen Codeteilen suchen.
mfg
Gerold
