Data Storage and Integration - GCP Professional Cloud Developer¶

from google.cloud import storage
from google.cloud.exceptions import GoogleCloudError
import os
from datetime import timedelta

class CloudStorageClient:
    def __init__(self, project_id=None):
        """Initialize Cloud Storage client with optional project ID"""
        self.client = storage.Client(project=project_id)

    def upload_file(self, bucket_name, source_file_name, destination_blob_name):
        """Upload file to bucket with error handling"""
        try:
            bucket = self.client.bucket(bucket_name)
            blob = bucket.blob(destination_blob_name)

            # Upload with metadata
            blob.metadata = {
                'uploaded_by': 'app-service',
                'original_name': os.path.basename(source_file_name)
            }
            blob.upload_from_filename(source_file_name)

            print(f"File {source_file_name} uploaded to {destination_blob_name}")
            return blob.public_url
        except GoogleCloudError as e:
            print(f"Error uploading file: {e}")
            raise

    def upload_from_memory(self, bucket_name, data, destination_blob_name, content_type='application/octet-stream'):
        """Upload data from memory"""
        bucket = self.client.bucket(bucket_name)
        blob = bucket.blob(destination_blob_name)
        blob.upload_from_string(data, content_type=content_type)
        return blob.public_url

    def download_file(self, bucket_name, source_blob_name, destination_file_name):
        """Download file from bucket"""
        try:
            bucket = self.client.bucket(bucket_name)
            blob = bucket.blob(source_blob_name)
            blob.download_to_filename(destination_file_name)
            print(f"Downloaded {source_blob_name} to {destination_file_name}")
        except GoogleCloudError as e:
            print(f"Error downloading file: {e}")
            raise

    def download_to_memory(self, bucket_name, source_blob_name):
        """Download file contents to memory"""
        bucket = self.client.bucket(bucket_name)
        blob = bucket.blob(source_blob_name)
        return blob.download_as_bytes()

    def list_blobs(self, bucket_name, prefix=None):
        """List all blobs in bucket with optional prefix filter"""
        bucket = self.client.bucket(bucket_name)
        blobs = bucket.list_blobs(prefix=prefix)
        return [blob.name for blob in blobs]

    def delete_blob(self, bucket_name, blob_name):
        """Delete a blob from bucket"""
        bucket = self.client.bucket(bucket_name)
        blob = bucket.blob(blob_name)
        blob.delete()
        print(f"Blob {blob_name} deleted")

from google.cloud import storage
from datetime import datetime, timedelta

def generate_signed_url(bucket_name, blob_name, expiration_minutes=60, method='GET'):
    """Generate signed URL for temporary access"""
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(blob_name)

    # Generate signed URL valid for specified minutes
    url = blob.generate_signed_url(
        version='v4',
        expiration=timedelta(minutes=expiration_minutes),
        method=method,
        # Optional: specify content type for uploads
        content_type='application/octet-stream' if method == 'PUT' else None
    )

    return url

def generate_upload_url(bucket_name, blob_name, content_type='application/octet-stream'):
    """Generate signed URL for file upload"""
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(blob_name)

    url = blob.generate_signed_url(
        version='v4',
        expiration=timedelta(hours=1),
        method='PUT',
        content_type=content_type
    )

    return url

# Usage example
download_url = generate_signed_url('my-bucket', 'documents/report.pdf', expiration_minutes=30)
upload_url = generate_upload_url('my-bucket', 'uploads/user-123/photo.jpg', 'image/jpeg')

from google.cloud import storage
from google.resumable_media import requests, common
import requests as standard_requests

def resumable_upload_large_file(bucket_name, source_file_name, destination_blob_name):
    """Upload large file with resumable upload"""
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(destination_blob_name)

    # Configure resumable upload with chunk size
    chunk_size = 5 * 1024 * 1024  # 5 MB chunks
    blob.chunk_size = chunk_size

    # Upload with progress tracking
    with open(source_file_name, 'rb') as file_obj:
        total_bytes = os.path.getsize(source_file_name)
        bytes_uploaded = 0

        # Custom upload with retry
        blob.upload_from_file(
            file_obj,
            rewind=True,
            size=total_bytes,
            num_retries=3
        )

    print(f"File {source_file_name} uploaded with resumable upload")

def resumable_upload_with_progress(bucket_name, source_file_name, destination_blob_name):
    """Resumable upload with progress callback"""
    from google.cloud.storage import Blob
    from google.cloud.storage.blob import _DEFAULT_CHUNKSIZE

    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(destination_blob_name)

    # Set custom chunk size (must be multiple of 256 KB)
    blob.chunk_size = 5 * 1024 * 1024  # 5 MB

    file_size = os.path.getsize(source_file_name)

    with open(source_file_name, 'rb') as file_obj:
        # Create a custom upload session
        blob.upload_from_file(
            file_obj,
            size=file_size,
            num_retries=5,
            if_generation_match=None
        )

    print(f"Upload completed: {destination_blob_name}")

from google.cloud import storage
import os
import math
from concurrent.futures import ThreadPoolExecutor, as_completed

def parallel_composite_upload(bucket_name, source_file_name, destination_blob_name, chunk_size=32*1024*1024):
    """Upload large file using parallel composite uploads"""
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)

    # Get file size
    file_size = os.path.getsize(source_file_name)
    num_chunks = math.ceil(file_size / chunk_size)

    # Upload chunks in parallel
    chunk_blobs = []

    def upload_chunk(chunk_num, start, end):
        chunk_blob_name = f"{destination_blob_name}_chunk_{chunk_num}"
        blob = bucket.blob(chunk_blob_name)

        with open(source_file_name, 'rb') as f:
            f.seek(start)
            chunk_data = f.read(end - start)
            blob.upload_from_string(chunk_data)

        return blob

    with ThreadPoolExecutor(max_workers=8) as executor:
        futures = []
        for i in range(num_chunks):
            start = i * chunk_size
            end = min((i + 1) * chunk_size, file_size)
            future = executor.submit(upload_chunk, i, start, end)
            futures.append(future)

        for future in as_completed(futures):
            chunk_blobs.append(future.result())

    # Compose chunks into final blob
    destination_blob = bucket.blob(destination_blob_name)
    destination_blob.compose(chunk_blobs)

    # Clean up chunk blobs
    for blob in chunk_blobs:
        blob.delete()

    print(f"Parallel upload completed: {destination_blob_name}")

def parallel_upload_large_file(bucket_name, source_file_name, destination_blob_name):
    """Simplified parallel upload for files > 100MB"""
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(destination_blob_name)

    # Cloud Storage handles parallel upload automatically for large files
    # Just configure appropriate chunk size
    blob.chunk_size = 10 * 1024 * 1024  # 10 MB chunks

    blob.upload_from_filename(source_file_name)
    print(f"Large file uploaded: {destination_blob_name}")

from google.cloud import storage
import io

def stream_upload(bucket_name, destination_blob_name, data_generator):
    """Upload data from a generator/stream"""
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(destination_blob_name)

    # Upload from file-like object
    with blob.open('wb') as f:
        for chunk in data_generator:
            f.write(chunk)

    print(f"Stream upload completed: {destination_blob_name}")

def stream_download(bucket_name, source_blob_name, process_chunk):
    """Download and process data as stream"""
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(source_blob_name)

    # Download as stream
    with blob.open('rb') as f:
        while True:
            chunk = f.read(1024 * 1024)  # 1 MB chunks
            if not chunk:
                break
            process_chunk(chunk)

    print(f"Stream download completed: {source_blob_name}")

# Example: Stream CSV processing
def process_csv_stream(bucket_name, blob_name):
    """Process CSV file without loading into memory"""
    import csv

    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(blob_name)

    with blob.open('r') as f:
        csv_reader = csv.DictReader(f)
        for row in csv_reader:
            # Process each row
            process_row(row)

def process_row(row):
    """Process individual CSV row"""
    print(f"Processing: {row}")

import sqlalchemy
from google.cloud.sql.connector import Connector
import pg8000

def create_connection_pool_with_connector():
    """Create connection pool using Cloud SQL Python Connector"""
    connector = Connector()

    def getconn():
        conn = connector.connect(
            "project:region:instance",
            "pg8000",
            user="db-user",
            password="db-password",
            db="database-name"
        )
        return conn

    # Create SQLAlchemy engine
    pool = sqlalchemy.create_engine(
        "postgresql+pg8000://",
        creator=getconn,
        pool_size=5,
        max_overflow=2,
        pool_timeout=30,
        pool_recycle=1800,
        pool_pre_ping=True  # Verify connections before using
    )

    return pool, connector

# Usage
pool, connector = create_connection_pool_with_connector()

try:
    with pool.connect() as conn:
        result = conn.execute(sqlalchemy.text("SELECT NOW()"))
        print(result.fetchone())
finally:
    connector.close()

import sqlalchemy
import os

def create_unix_socket_pool():
    """Create connection pool using Unix socket (App Engine)"""
    db_user = os.environ.get("DB_USER")
    db_pass = os.environ.get("DB_PASS")
    db_name = os.environ.get("DB_NAME")
    instance_connection_name = os.environ.get("INSTANCE_CONNECTION_NAME")

    db_socket_dir = "/cloudsql"

    pool = sqlalchemy.create_engine(
        sqlalchemy.engine.url.URL.create(
            drivername="postgresql+pg8000",
            username=db_user,
            password=db_pass,
            database=db_name,
            query={"unix_sock": f"{db_socket_dir}/{instance_connection_name}/.s.PGSQL.5432"}
        ),
        pool_size=5,
        max_overflow=2,
        pool_timeout=30,
        pool_recycle=1800,
        pool_pre_ping=True
    )

    return pool

import sqlalchemy
import os

def create_private_ip_pool():
    """Create connection pool using private IP (VPC)"""
    db_user = os.environ.get("DB_USER")
    db_pass = os.environ.get("DB_PASS")
    db_name = os.environ.get("DB_NAME")
    db_host = os.environ.get("PRIVATE_IP")  # Private IP of Cloud SQL instance

    pool = sqlalchemy.create_engine(
        sqlalchemy.engine.url.URL.create(
            drivername="postgresql+pg8000",
            username=db_user,
            password=db_pass,
            host=db_host,
            port=5432,
            database=db_name
        ),
        pool_size=5,
        max_overflow=2,
        pool_timeout=30,
        pool_recycle=1800,
        pool_pre_ping=True
    )

    return pool

import sqlalchemy
from sqlalchemy.pool import NullPool, QueuePool
from google.cloud.sql.connector import Connector
import logging

class DatabaseConnection:
    def __init__(self):
        self.connector = Connector()
        self.engine = self._create_engine()

    def _create_engine(self):
        """Create optimized database engine with connection pooling"""
        def getconn():
            conn = self.connector.connect(
                "project:region:instance",
                "pg8000",
                user=os.environ.get("DB_USER"),
                password=os.environ.get("DB_PASS"),
                db=os.environ.get("DB_NAME"),
                # Enable faster IAM authentication
                enable_iam_auth=False
            )
            return conn

        engine = sqlalchemy.create_engine(
            "postgresql+pg8000://",
            creator=getconn,
            # Pool configuration for Cloud Run/Functions
            pool_size=5,  # Core pool size
            max_overflow=2,  # Additional connections when pool is full
            pool_timeout=30,  # Timeout waiting for connection
            pool_recycle=1800,  # Recycle connections after 30 min
            pool_pre_ping=True,  # Verify connection before using
            echo=False,  # Set to True for SQL debugging
            pool_use_lifo=True  # Use LIFO for better connection reuse
        )

        return engine

    def get_connection(self):
        """Get database connection from pool"""
        return self.engine.connect()

    def execute_query(self, query, params=None):
        """Execute query with automatic connection management"""
        with self.engine.connect() as conn:
            result = conn.execute(sqlalchemy.text(query), params or {})
            return result.fetchall()

    def execute_transaction(self, operations):
        """Execute multiple operations in a transaction"""
        with self.engine.begin() as conn:
            results = []
            for operation in operations:
                result = conn.execute(
                    sqlalchemy.text(operation['query']),
                    operation.get('params', {})
                )
                results.append(result)
            return results

    def close(self):
        """Close all connections and cleanup"""
        self.engine.dispose()
        self.connector.close()

# Usage
db = DatabaseConnection()

try:
    # Simple query
    users = db.execute_query("SELECT * FROM users WHERE active = :active", {"active": True})

    # Transaction
    operations = [
        {"query": "UPDATE accounts SET balance = balance - :amount WHERE id = :id",
         "params": {"amount": 100, "id": 1}},
        {"query": "UPDATE accounts SET balance = balance + :amount WHERE id = :id",
         "params": {"amount": 100, "id": 2}}
    ]
    db.execute_transaction(operations)
finally:
    db.close()

from sqlalchemy import create_engine, Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from datetime import datetime
from google.cloud.sql.connector import Connector

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    email = Column(String(255), unique=True, nullable=False)
    name = Column(String(100))
    created_at = Column(DateTime, default=datetime.utcnow)
    orders = relationship("Order", back_populates="user")

class Order(Base):
    __tablename__ = 'orders'

    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id'))
    total_amount = Column(Integer)
    status = Column(String(50))
    created_at = Column(DateTime, default=datetime.utcnow)
    user = relationship("User", back_populates="orders")

# Database setup
connector = Connector()

def getconn():
    return connector.connect(
        "project:region:instance",
        "pg8000",
        user=os.environ["DB_USER"],
        password=os.environ["DB_PASS"],
        db=os.environ["DB_NAME"]
    )

engine = create_engine(
    "postgresql+pg8000://",
    creator=getconn,
    pool_pre_ping=True,
    pool_size=5,
    max_overflow=2
)

Session = sessionmaker(bind=engine)

# CRUD Operations
def create_user(email, name):
    """Create new user"""
    session = Session()
    try:
        user = User(email=email, name=name)
        session.add(user)
        session.commit()
        session.refresh(user)
        return user
    except Exception as e:
        session.rollback()
        raise
    finally:
        session.close()

def get_user_with_orders(user_id):
    """Get user with all orders"""
    session = Session()
    try:
        user = session.query(User).filter(User.id == user_id).first()
        if user:
            # Eagerly load orders
            orders = user.orders
            return {
                'id': user.id,
                'email': user.email,
                'name': user.name,
                'orders': [{'id': o.id, 'total': o.total_amount, 'status': o.status} for o in orders]
            }
        return None
    finally:
        session.close()

def create_order_with_transaction(user_id, total_amount):
    """Create order with transaction"""
    session = Session()
    try:
        # Verify user exists
        user = session.query(User).filter(User.id == user_id).first()
        if not user:
            raise ValueError("User not found")

        # Create order
        order = Order(user_id=user_id, total_amount=total_amount, status='pending')
        session.add(order)
        session.commit()
        return order
    except Exception as e:
        session.rollback()
        raise
    finally:
        session.close()

# alembic/env.py
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context
from google.cloud.sql.connector import Connector
import os

# Import your models
from myapp.models import Base

config = context.config

# Setup connector for migrations
connector = Connector()

def getconn():
    return connector.connect(
        os.environ["INSTANCE_CONNECTION_NAME"],
        "pg8000",
        user=os.environ["DB_USER"],
        password=os.environ["DB_PASS"],
        db=os.environ["DB_NAME"]
    )

def run_migrations_online():
    """Run migrations in 'online' mode."""
    configuration = config.get_section(config.config_ini_section)
    configuration["sqlalchemy.url"] = "postgresql+pg8000://"

    connectable = engine_from_config(
        configuration,
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
        creator=getconn
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=Base.metadata
        )

        with context.begin_transaction():
            context.run_migrations()

    connector.close()

run_migrations_online()

from google.cloud import spanner
from google.cloud.spanner_v1 import param_types

class SpannerClient:
    def __init__(self, instance_id, database_id):
        self.spanner_client = spanner.Client()
        self.instance = self.spanner_client.instance(instance_id)
        self.database = self.instance.database(database_id)

    def read_data(self, table, columns, key_set):
        """Read data using primary key"""
        with self.database.snapshot() as snapshot:
            results = snapshot.read(
                table=table,
                columns=columns,
                keyset=key_set
            )
            return [dict(zip(columns, row)) for row in results]

    def query_data(self, query, params=None, param_types_dict=None):
        """Execute SQL query"""
        with self.database.snapshot() as snapshot:
            results = snapshot.execute_sql(
                query,
                params=params,
                param_types=param_types_dict
            )
            return [dict(row) for row in results]

    def insert_data(self, table, columns, values):
        """Insert data using mutation"""
        with self.database.batch() as batch:
            batch.insert(
                table=table,
                columns=columns,
                values=values
            )

    def update_data(self, table, columns, values):
        """Update data using mutation"""
        with self.database.batch() as batch:
            batch.update(
                table=table,
                columns=columns,
                values=values
            )

    def delete_data(self, table, key_set):
        """Delete data using mutation"""
        with self.database.batch() as batch:
            batch.delete(table=table, keyset=key_set)

from google.cloud import spanner

def create_spanner_database_with_pool(instance_id, database_id):
    """Create Spanner database with optimized session pool"""
    spanner_client = spanner.Client()
    instance = spanner_client.instance(instance_id)

    # Configure session pool
    pool = spanner.BurstyPool(
        target_size=10,  # Target number of sessions
        max_size=30,  # Maximum sessions
        min_size=5  # Minimum sessions to maintain
    )

    database = instance.database(
        database_id,
        pool=pool
    )

    return database

# Usage with session management
database = create_spanner_database_with_pool('my-instance', 'my-database')

# Sessions are automatically managed by the pool
def execute_query_with_session():
    with database.snapshot() as snapshot:
        results = snapshot.execute_sql(
            "SELECT * FROM Users WHERE active = @active",
            params={"active": True},
            param_types={"active": param_types.BOOL}
        )
        return list(results)

from google.cloud import spanner

database = instance.database('my-database')

def transfer_funds(from_account_id, to_account_id, amount):
    """Transfer funds between accounts with transaction"""
    def update_in_transaction(transaction):
        # Read current balances
        row = list(transaction.read(
            table='Accounts',
            columns=['account_id', 'balance'],
            keyset=spanner.KeySet(keys=[[from_account_id]])
        ))[0]

        from_balance = row[1]

        if from_balance < amount:
            raise ValueError("Insufficient funds")

        # Update balances
        transaction.update(
            table='Accounts',
            columns=['account_id', 'balance'],
            values=[
                [from_account_id, from_balance - amount],
                [to_account_id, from_balance + amount]  # Simplified - should read to_account balance
            ]
        )

    database.run_in_transaction(update_in_transaction)

def batch_update_with_retry(updates):
    """Batch update with automatic retry"""
    from google.api_core import retry

    @retry.Retry(predicate=retry.if_transactional_error)
    def update_in_transaction(transaction):
        for update in updates:
            transaction.update(
                table=update['table'],
                columns=update['columns'],
                values=update['values']
            )

    database.run_in_transaction(update_in_transaction)