Przejdź do głównej zawartości

PostgreSQL

Jeśli dotarłeś tak daleko, wierzymy, że już wiesz, jak stworzyć bazę danych w naszej aplikacji CLI. Jeśli nie, sprawdź sekcję Bazy danych i utwórz jedną do dalszego szkolenia.

Przygotowanie​

Dla PostgreSQL, nasz wrapper został wzbogacony o przydatne polecenia do interakcji z DB. Po pierwsze, tworzymy połączenie z DB za pomocą:

  • nazwy działającej instancji psql
  • wygenerowanego dla tej instancji tokena

Jeśli nie masz żadnych działających instancji PostgreSQL, utwórz jedną, a następnie skopiuj nazwę i token aplikacji.
Token aplikacji można znaleźć, korzystając z:

cgc db list -d

To nie zadziała z Jupyterem, chyba że dostarczyłeś klucze API. Użyj swojego osobistego terminala, gdzie dokonałeś rejestracji w cgc.

Zrozumienie PostgreSQL w CGC​

Czym jest PostgreSQL?​

Dla użytkowników nietechnicznych:

PostgreSQL jest jak cyfrowa szafa na dokumenty:

  • Przechowuje dane w zorganizowanych tabelach (jak arkusze kalkulacyjne)
  • Możesz szybko wyszukiwać, aktualizować i pobierać informacje
  • Wiele aplikacji może współdzielić te same dane
  • Zapewnia bezpieczeństwo i spójność danych

Dla użytkowników technicznych:

PostgreSQL w CGC:

  • Działa jako usługa kontenerowa w Kubernetes
  • Obsługuje standardowe funkcje PostgreSQL (wersja 17+)
  • Wykorzystuje odkrywanie usług Kubernetes do łączności
  • Wymaga natywnych klientów PostgreSQL w Pythonie (psycopg2)
  • Dostęp sieciowy ograniczony do przestrzeni nazw Kubernetes

Wymagania​

Przed rozpoczęciem pracy z PostgreSQL upewnij się, że masz zainstalowany wymagany pakiet Pythona:

pip install psycopg2-binary

Rozpoczęcie pracy​

Importowanie wymaganych modułów​

import psycopg2
from psycopg2 import errors as pg_errors
import cgc.sdk.resource as resource
import cgc.sdk.exceptions as exceptions

Wdrażanie bazy danych PostgreSQL​

def deploy_postgres(db_name="example-postgres", password="example-pass-123"):
"""Wdrażanie zasobu bazy danych PostgreSQL"""
    response = resource.resource_create(
name=db_name,
image_name="postgres:17",
entity="postgresql",
cpu=2,
memory=4,
environment_data=[
f"POSTGRES_PASSWORD={password}",
"POSTGRES_USER=admin",
"POSTGRES_DB=db"
        ]
    )
if response['code'] == 200:
print(f"✓ Baza danych PostgreSQL '{db_name}' utworzona pomyślnie")
# Oczekiwanie na gotowość bazy danych
while not resource.resource_ready(db_name, resource.ResourceTypes.db):
            time.sleep(5)
return db_name, password
else:
print(f"✗ Nie udało się utworzyć bazy danych: {response.get('message')}")
return None, None

Połączenie z PostgreSQL​

def connect_to_database(db_name, password, database="db"):
"""Łączenie z bazą danych PostgreSQL za pomocą natywnego psycopg2"""
# Parametry połączenia dla usługi Kubernetes
    connection_params = {
'host': db_name, # Nazwa usługi Kubernetes
'port': 5432, # Standardowy port PostgreSQL
'user': 'admin', # Użytkownik bazy danych
'password': password, # Hasło bazy danych
'database': database, # Nazwa bazy danych
'connect_timeout': 10
    }
# Połączenie z logiką ponawiania
    max_retries = 5
for attempt in range(max_retries):
try:
            connection = psycopg2.connect(**connection_params)
# Testowanie połączenia
            cursor = connection.cursor()
            cursor.execute("SELECT version()")
            version = cursor.fetchone()[0]
            cursor.close()
print(f"✓ Połączono z PostgreSQL: {version[:50]}...")
return connection
except Exception as e:
print(f"Próba połączenia {attempt + 1} nieudana: {e}")
if attempt < max_retries - 1:
                time.sleep(10)
else:
raise
return None

Praca z bazami danych​

Tworzenie tabel​

def create_tables(connection):
"""Tworzenie przykładowych tabel"""
    cursor = connection.cursor()
try:
# Tworzenie tabeli użytkowników
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id SERIAL PRIMARY KEY,
                username VARCHAR(50) UNIQUE NOT NULL,
                email VARCHAR(100) UNIQUE NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
# Tworzenie tabeli postów
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS posts (
                id SERIAL PRIMARY KEY,
                user_id INTEGER REFERENCES users(id),
                title VARCHAR(200) NOT NULL,
                content TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        connection.commit()
print("✓ Tabele utworzone pomyślnie")
return True
except Exception as e:
        connection.rollback()
print(f"✗ Błąd podczas tworzenia tabel: {e}")
return False
finally:
        cursor.close()

Wstawianie danych​

def insert_sample_data(connection):
"""Wstawianie przykładowych danych do tabel"""
    cursor = connection.cursor()
try:
# Wstawianie użytkowników
        users = [
            ("alice", "alice@example.com"),
            ("bob", "bob@example.com"),
            ("charlie", "charlie@example.com")
        ]
        user_ids = []
for username, email in users:
            cursor.execute(
"INSERT INTO users (username, email) VALUES (%s, %s) RETURNING id",
                (username, email)
            )
            user_id = cursor.fetchone()[0]
            user_ids.append(user_id)
# Wstawianie postów
        posts = [
            (user_ids[0], "Witaj, świecie", "To mój pierwszy post!"),
            (user_ids[1], "Przykład bazy danych", "PostgreSQL jest świetny!"),
        ]
for user_id, title, content in posts:
            cursor.execute(
"INSERT INTO posts (user_id, title, content) VALUES (%s, %s, %s)",
                (user_id, title, content)
            )
        connection.commit()
print(f"✓ Wstawiono {len(users)} użytkowników i {len(posts)} postów")
return True
except pg_errors.UniqueViolation:
        connection.rollback()
print("Uwaga: Niektóre dane już istnieją (duplikaty wpisów)")
return True
except Exception as e:
        connection.rollback()
print(f"✗ Błąd podczas wstawiania danych: {e}")
return False
finally:
        cursor.close()

Zapytania do danych​

def query_data(connection):
"""Wykonywanie zapytań i wyświetlanie danych"""
    cursor = connection.cursor()
try:
# Liczenie użytkowników
        cursor.execute("SELECT COUNT(*) FROM users")
        user_count = cursor.fetchone()[0]
print(f"Łączna liczba użytkowników: {user_count}")
# Lista użytkowników z liczbą postów
        cursor.execute("""
            SELECT
                u.username,
                u.email,
                COUNT(p.id) as post_count
            FROM users u
            LEFT JOIN posts p ON u.id = p.user_id
            GROUP BY u.id, u.username, u.email
            ORDER BY post_count DESC
        """)
for row in cursor.fetchall():
            username, email, post_count = row
print(f"Użytkownik: {username} ({email}) - {post_count} postów")
return True
except Exception as e:
print(f"✗ Błąd podczas wykonywania zapytań: {e}")
return False
finally:
        cursor.close()

Zaawansowane użytkowanie​

Zarządzanie transakcjami​

def demonstrate_transactions(connection):
"""Demonstracja obsługi transakcji"""
    cursor = connection.cursor()
try:
# psycopg2 automatycznie rozpoczyna transakcje
print("Rozpoczynanie transakcji...")
# Operacja 1: Tworzenie nowego użytkownika
        cursor.execute(
"INSERT INTO users (username, email) VALUES (%s, %s) RETURNING id",
            ("transaction_test", "trans@example.com")
        )
        new_user_id = cursor.fetchone()[0]
print(f"✓ Utworzono użytkownika (ID: {new_user_id})")
# Operacja 2: Tworzenie postów
for i in range(3):
            cursor.execute(
"INSERT INTO posts (user_id, title, content) VALUES (%s, %s, %s)",
                (new_user_id, f"Post transakcji {i+1}", "Testowanie transakcji")
            )
print("✓ Utworzono 3 posty")
# Zatwierdzenie transakcji
        connection.commit()
print("✓ Transakcja zatwierdzona pomyślnie")
# Demonstracja wycofania (celowy błąd)
try:
            cursor.execute(
"INSERT INTO users (username, email) VALUES (%s, %s)",
                ("transaction_test", "another@example.com") # Zduplikowana nazwa użytkownika
            )
            connection.commit()
except pg_errors.UniqueViolation:
            connection.rollback()
print("✓ Transakcja wycofana (zduplikowana nazwa użytkownika)")
return True
except Exception as e:
        connection.rollback()
print(f"✗ Transakcja nieudana: {e}")
return False
finally:
        cursor.close()

Zarządzanie połączeniem za pomocą menedżera kontekstu​

import contextlib
@contextlib.contextmanager
def database_connection(db_name, password, database="db"):
"""Menedżer kontekstu dla połączeń z bazą danych"""
    connection = None
try:
# Połączenie z bazą danych
        connection = psycopg2.connect(
host=db_name,
port=5432,
user='admin',
password=password,
database=database,
connect_timeout=10
        )
yield connection
except Exception as e:
if connection:
            connection.rollback()
raise e
finally:
if connection:
            connection.close()
# Użycie
with database_connection("postgres-db", "password") as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT COUNT(*) FROM users")
    count = cursor.fetchone()[0]
print(f"Łączna liczba użytkowników: {count}")
    cursor.close()

Kompletny przykład​

Oto kompletny przykład z przykładów SDK:

#!/usr/bin/env python3
"""
Przykład połączenia z bazą danych PostgreSQL
Wymagania:
    pip install psycopg2-binary
"""
import time
import psycopg2
from psycopg2 import errors as pg_errors
import cgc.sdk.resource as resource
import cgc.sdk.exceptions as exceptions
def main():
"""Główny przepływ wykonania"""
print("CGC SDK - Przykład bazy danych PostgreSQL (natywny psycopg2)")
# Krok 1: Wdrożenie PostgreSQL
    db_name, password = deploy_postgres()
if not db_name:
print("Nie udało się wdrożyć bazy danych. Kończenie.")
return
# Potwierdzenie dostępu sieciowego
print("\\nPOTWIERDZENIE DOSTĘPU SIECIOWEGO")
print("Dostęp do bazy danych jest dostępny tylko w obrębie sieci przestrzeni nazw Kubernetes.")
    confirmation = input("Czy obecnie znajdujesz się w sieci przestrzeni nazw? (y/N): ")
if confirmation.lower().strip() != 'y':
print("✗ Dostęp sieciowy niepotwierdzony. Przetwarzanie czyszczenia...")
cleanup_database(db_name)
return
# Krok 2: Połączenie z bazą danych
    connection = connect_to_database(db_name, password)
if not connection:
print("Nie udało się połączyć z bazą danych. Czyszczenie...")
cleanup_database(db_name)
return
try:
# Krok 3: Tworzenie tabel i praca z danymi
create_tables(connection)
insert_sample_data(connection)
query_data(connection)
demonstrate_transactions(connection)
finally:
# Zawsze zamykaj połączenie
        connection.close()
print("✓ Połączenie z bazą danych zamknięte")
# Opcja czyszczenia
    user_input = input(f"Czy chcesz usunąć bazę danych '{db_name}'? (y/n): ")
if user_input.lower() == 'y':
cleanup_database(db_name)
if __name__ == "__main__":
main()

Wymagania sieciowe​

Ważne: Bazy danych PostgreSQL wdrożone za pomocą CGC SDK są dostępne tylko w obrębie sieci przestrzeni nazw Kubernetes. Ze względów bezpieczeństwa NIE są one dostępne przez ingress. Aby połączyć się z bazą danych, musisz:

  1. Uruchamiać kod w tej samej przestrzeni nazw Kubernetes
  2. Używać nazwy usługi (db_name) jako nazwy hosta
  3. Używać portu 5432 (standardowy port PostgreSQL)

Najlepsze praktyki​

1. Bezpieczeństwo​

  • Nigdy nie wpisuj haseł bezpośrednio w kodzie
  • Używaj zmiennych środowiskowych lub bezpiecznych monitów o hasło
  • Zawsze stosuj parametryzowane zapytania, aby zapobiec atakom SQL injection

2. Obsługa błędów​

from psycopg2 import errors as pg_errors
try:
# Operacje na bazie danych
    cursor.execute("INSERT INTO users ...")
    connection.commit()
except pg_errors.UniqueViolation:
    connection.rollback()
print("Użytkownik już istnieje")
except pg_errors.ConnectionException:
print("Utracono połączenie z bazą danych")
except Exception as e:
    connection.rollback()
print(f"Nieoczekiwany błąd: {e}")

3. Zarządzanie połączeniami​

  • Zawsze zamykaj kursory i połączenia
  • Używaj menedżerów kontekstu, gdy to możliwe
  • Wdrażaj logikę ponawiania dla nieudanych połączeń
  • Ponownie używaj połączeń zamiast tworzenia nowych za każdym razem

4. Wydajność​

  • Używaj puli połączeń dla aplikacji o dużym ruchu
  • Dodawaj indeksy do często wyszukiwanych kolumn
  • Używaj operacji masowych do wstawiania wielu rekordów
  • Monitoruj wydajność zapytań za pomocą EXPLAIN ANALYZE

Rozwiązywanie problemów​

Problemy z połączeniem​

Problem: "nie można przetłumaczyć nazwy hosta na adres"

Rozwiązania:

  1. Upewnij się, że działasz w obrębie przestrzeni nazw Kubernetes
  2. Sprawdź, czy zasób bazy danych jest wdrożony i gotowy
  3. Upewnij się, że nazwa usługi odpowiada nazwie bazy danych

Problem: "Odmowa połączenia"

Rozwiązania:

  1. Sprawdź, czy baza danych działa: resource.resource_ready(db_name, resource.ResourceTypes.db)
  2. Upewnij się, że baza danych zakończyła inicjalizację (może to zająć 30-60 sekund)
  3. Upewnij się, że używasz prawidłowego portu (5432)

Problem: "Nieudane uwierzytelnianie"

Rozwiązania:

  1. Sprawdź, czy hasło jest poprawne
  2. Upewnij się, że nazwa użytkownika to 'admin' (domyślna dla CGC PostgreSQL)
  3. Sprawdź, czy nazwa bazy danych istnieje

Migracja z klienta PostgreSQL SDK​

Jeśli wcześniej używałeś cgc.sdk.postgresql, oto kluczowe zmiany:

Stary sposób:

import cgc.sdk.postgresql as pg
db_connector = pg.postgresql_client(app_name, password, database)
connection = db_connector.get_postgresql_client()

Nowy sposób:

import psycopg2
connection = psycopg2.connect(
host=app_name,
port=5432,
user='admin',
password=password,
database=database
)