Kategorie: Python

Um mit einer PostgreSQL Datenbank eine Verbindung aufzubauen, Daten auszulesen, dafür kann man folgendes Skript verwenden:

Zuerst einmal installieren wir psycopg:

sudo python -m pip install psycopg
Oder suso pip install psycopg

Dann bereiten wir die Datenbank „sample“ vor:

DROP TABLE IF EXISTS `sample`;
CREATE TABLE IF NOT EXISTS `sample` (
  `id` int NOT NULL AUTO_INCREMENT,
  `record_no` int NOT NULL,
  `description` varchar(50) NOT NULL,
  `start_date` datetime NOT NULL,
  PRIMARY KEY (`id`),
  KEY `record_no_key` (`record_no`)
) ENGINE=MyISAM DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
COMMIT;

Dann das Script:

sudo nano psycopg_test.py

import psycopg
from psycopg import connect as pg_connect, rows as pg_rows

connection = None

try:
    connection = pg_connect(
            user="postgres",
            password="admin",
            host="127.0.0.1",
            port="5432",
            dbname="test_py",
            row_factory=pg_rows.namedtuple_row
    )
    cursor = connection.cursor()
except (Exception, psycopg.Error) as error:
    print("Error while connecting to PostgreSQL", error)

select_Query = 'SELECT * FROM sample'

cursor.execute(select_Query)

samples = cursor.fetchall()

for sample in samples:
    print("description = ", sample.description, "\n")

if connection:
    cursor.close()
    connection.close()
    print("PostgreSQK connection is closed")

Wir verwenden nur psycopg, also genau genommen psycopg3.

Bitte das Script nicht psycopg.py benennen, da Python den Dateinamen als Objekt erachtet und es natürlich psycopg nicht findet!

sudo pythob psycopg_test.py

Getestet unter CentOs 7 und Debian 11 und Windows 10

Wie immer, alle Angaben ohne Gewähr, Anwendung auf eigene Gefahr und Verantwortung, ich übernehme keinerlei Haftung für Ausfälle, Datenverlust oder andere finanzielle Schäden.

Happy coding!

Um MySQL mit dem MySql Connector zu verwenden muss dieser zuerst in Python installiert werden:

sudo python -m pip install mysql-connector-python
oder sudo pip install mysql-connector-python
oder pip install mysql-connector-python
oder pip3 install mysql-connector-python

Für Windows halt ohne "sudo", ich empfehle: führen Sie CMD immer als Administrator aus, das geht mit WIN+R, dann cmd.exe eingeben und Shift+Strg+Enter drücken, dann startet man die CMD "elevated" also als Administrator. Dann bleibt man nirgendwo stecken, nur weil man keine Rechte hat...

Dann legen wir eine neue Datenbank am lokalen Computer/Server an:

DROP TABLE IF EXISTS `sample`;
CREATE TABLE IF NOT EXISTS `sample` (
  `id` int NOT NULL AUTO_INCREMENT,
  `record_no` int NOT NULL,
  `description` varchar(50) NOT NULL,
  `start_date` datetime NOT NULL,
  PRIMARY KEY (`id`),
  KEY `record_no_key` (`record_no`)
) ENGINE=MyISAM DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
COMMIT;

Einfach mit PhpMyAdmin auf die lokale Datenbank verbinden und das .SQL Script ausführen, natürlich kann man noch das Script als .sql Datei speichern und dann per „Importieren“ dann ausführen, bei kleineren Dateien geht Copy/Paste bei größeren Importieren, wenn die Datei riesig ist, muss man direkt die Datei importieren (z.B. mit mysql -u username -p database_name < sample.sql)

Dann erstellen wir unser Pythonscript:

(Anmerkung: das Script NICHT mysql.py benennen, sonst überschreiben wir das Modul mysql mit unserem und das mysql Modul wird nicht gefunden!)

sudo nano mysql_test.py

Der Inhalt:

import mysql.connector
from datetime import date

cn_connect = mysql.connector.connect(
    host='localhost',
    user='root',
    password='admin',
    database='test_py'
)

# connect to mysql, cursor is our "recordset"
cursor = cn_connect.cursor()

# date of today
datum = date.today()

try:
    # select last record_no
    sql_query = "SELECT * FROM sample ORDER BY record_no DESC LIMIT 1 OFFSET 0"
    cursor.execute(sql_query)
    
    # fetch all records, even only one is in the buffer
    record = cursor.fetchone()
    
    # if we got a record, then get it
    if record:
        # read last_record_no
        last_record_no=record[1]
    else:
        # otherwise last_record_no=0
        last_record_no=0    
except mysql.connector.Error as error:
    print("Error while connecting to MySQL", error)

# prepare sql query for insert
sql_insert = "INSERT INTO sample (record_no, description, start_date) VALUES (%s, %s, %s)"
data = (last_record_no+1, "*Automatic insert*", datum)

# add record to database
cursor.execute(sql_insert, data)

# finish transaction = important!
cn_connect.commit()

# close cursor
cursor.close()

# close connection
cn_connect.close()

Ausführen:

sudo python mysql_test.py

Unser Script liest den letzten Datensatz in der Tabelle „sample“, liest den letzten „record_no“ aus, später wird dieser wieder mit +1 eingefügt. Wenn kein Datensatz ursprünglich in der Tabelle war (wir testen mit if record:) dann wird die variable last_record_no mit 0 initialisiert. 0+1 ergibt dann den Start record_no als 1. Bitte nicht record_no als ID verwenden, das überlassen wir hübsch MySQL mit der Auto-ID, das funktioniert und hat mit der record_no nichts zu tun, die verwenden wir später für Relationen, Links usw. um z.B. 2 Datenbanken verbinden zu können oder sauber zu Synchronisieren, wir verlassen uns nicht auf die MySQL-ID.

Getestet unter CentOs 7, Debian 11 und diesmal auch Windows 10 (wird mit Windows 11 genau so funktionieren)

Wie immer, alle Angaben ohne Gewähr, Anwendung auf eigene Gefahr und Verantwortung, ich übernehme keinerlei Haftung für Ausfälle, Datenverlust oder andere finanzielle Schäden.

Happy coding!

So es geht weiter mit dem Blog, heute beschäftigen wir uns mit einer Flask Anwendung einer REST API. Ich verwende das selbst, um mit meinem laufenden Python Service zu kommunizieren.

Damit ist es einfach eine API mit Python zu erstellen. Diese wird immer per HTTP angesprochen, kann mit WSGI und Apache auch über SSL laufen. In diesem Test starten wir Flask aber als Service ohne SSL und direkt über einen Port, dieser sollte unterschiedlich zu Port 80, 443 usw. sein. Bitte nur zu Testzwecken Flask so betreiben.

Zuerst ändern wir den Code vom Python Debian Service, diesen haben wir bei https://www.predl.cc/python-debian-service/ erstellt.

Öffnen wir die my_service.py und ändern den Code auf:

from flask import Flask
app = Flask(__name__)
@app.route('/api', methods=['GET'])
def meine_aufgabe_route():
    # Hier kannst du zusätzlichen Code für den API-Zugriff schreiben
    return 'API-Antwort'
if __name__ == '__main__':
    app.run(port=5001)

Damit ist das Service fertig, es kann getestet werden, zuerst stoppen wir das alte Service:

sudo systemctl stop my_service.service

sudo systemctl start my_service.service

Dann sollte folgendes herauskommen:

 * Serving Flask app 'test'
 * Debug mode: off
WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI serr instead.
 * Running on http://127.0.0.1:5001
Press CTRL+C to quit

Um mit dem Service zu kommunizieren, öffnen wir eine weitere Shell, lassen diese aber geöffnet, sonst beendet sich unsere Flask Anwendung:

sudo wget http://127.0.0.1:5001/api
--2023-06-17 11:50:57--  http://127.0.0.1:5001/api
Connecting to 127.0.0.1:5001... connected.
HTTP request sent, awaiting response... 200 OK
Length: 11 [text/html]
Saving to: ‘api’

100%[===========================================================================================================================================>] 11          --.-K/s   in 0s

2023-06-17 11:50:57 (2.43 MB/s) - ‘api’ saved [11/11]

Wget erstellt die api Textdatei, in dieser ist die Antwort enthalten:

sudo nano api

Oder wir lassen uns den Inhalt gleich ausgeben:

sudo wget -q -O- http://127.0.0.1:5001/api
API-Antwort

Getestet unter CentOs 7 und Debian 11

Wie immer, alle Angaben ohne Gewähr, Anwendung auf eigene Gefahr und Verantwortung, ich übernehme keinerlei Haftung für Ausfälle, Datenverlust oder andere finanzielle Schäden.

Happy coding!

Wir starten heute unseren Blog mit einer Anleitung für ein Linux Python Service. Ein Programm dass ständig läuft und auch nach dem Neustart des Computers wieder gestartet wird:

Zuerst einmal den Python Code, der ständig laufen soll:

sudo nano my_service.py

Dann den Python Code:

import time

while True:
    print("Service läuft...")
    time.sleep(5)

Dann müssen wir noch das Service selbst erstellen, das ist einfacher als man denkt:

sudo nano /etc/systemd/system/my_service.service

Der Inhalt des Services:

[Unit]
Description=My Python Service
After=network.target

[Service]
ExecStart=/usr/bin/python3 /Pfad/zum/Python-Skript/my_service.py
WorkingDirectory=/Pfad/zum/Python-Skript/

[Install]
WantedBy=multi-user.target

So nun ist alles vorbereitet, wir können unser Service aktivieren, damit das Service auch nach einem Neustart wieder verfügbar ist:

sudo systemctl enable my_service.service

Das Service starten:

sudo systemctl start my_service.service

Überprüfen ob das Service läuft:

sudo systemctl status my_service.service

Und wieder beenden:

sudo systemctl stop my_service.service

Getestet unter CentOs 7 und Debian 11

Wie immer, alle Angaben ohne Gewähr, Anwendung auf eigene Gefahr und Verantwortung, ich übernehme keinerlei Haftung für Ausfälle, Datenverlust oder andere finanzielle Schäden.

Happy coding!