import faiss
import numpy as np

# 1. ベクトルデータの準備
# 例として、128次元のベクトルが1000個あると仮定
dimension = 128
num_vectors = 1000
data_vectors = np.random.rand(num_vectors, dimension).astype('float32')

# 2. インデックスの選択と初期化
# 最もシンプルなインデックス: IndexFlatL2
# L2 はユークリッド距離を意味し、最も厳密な（近似ではない）検索を行う
index = faiss.IndexFlatL2(dimension)

print(f"インデックスが空か？: {index.is_trained}") # IndexFlatL2は訓練不要なのでTrue
print(f"インデックスのベクトルの数: {index.ntotal}") # 初期状態では0

# 3. インデックスへのベクトルの追加
index.add(data_vectors)

print(f"インデックスのベクトルの数: {index.ntotal}") # 1000

# 1. クエリベクトルの準備
# 例として、1つのクエリベクトルを準備
query_vector = np.random.rand(1, dimension).astype('float32') # 1つのクエリは (1, dimension) 形状

# 2. 検索の実行
# k=5: 最も近い5つのベクトルを検索
k = 5
distances, indices = index.search(query_vector, k)

print(f"検索されたベクトルのインデックス (元の data_vectors での): {indices}")
print(f"検索されたベクトルの距離: {distances}")

# インデックスの保存
faiss.write_index(index, "my_faiss_index.bin")
print("インデックスを 'my_faiss_index.bin' に保存しました。")

# インデックスのロード
loaded_index = faiss.read_index("my_faiss_index.bin")
print(f"ロードされたインデックスのベクトルの数: {loaded_index.ntotal}")

# test_calculations.py
from my_functions import add, subtract, multiply

def test_add_positive_numbers():
    assert add(2, 3) == 5

def test_add_negative_numbers():
    assert add(-1, -5) == -6

@pytest.fixture(scope="module") # モジュールスコープに設定
def module_db_connection():
    """モジュール内で一度だけ実行されるDB接続のフィクスチャ"""
    print("\n[Module Scope] DB接続を確立しました。")
    db_conn = {"status": "connected", "data": []}
    yield db_conn
    print("[Module Scope] DB接続を切断しました。")

@pytest.fixture
def empty_db():
    """A fixture that provides an empty dictionary, simulating an empty database."""
    print("\nSetting up empty_db...") # This will print during test execution
    _mock_db.clear() # Ensure it's empty before each test using this fixture
    yield _mock_db # Yield the resource to the test
    print("Tearing down empty_db...") # This runs after the test finishes
    _mock_db.clear() # Clean up after the test

@pytest.fixture
def populated_db():
    """A fixture that provides a populated dictionary, simulating a database with some data."""
    print("\nSetting up populated_db...")
    _mock_db.clear()
    _mock_db["user1"] = {"name": "Alice", "email": "alice@example.com"}
    _mock_db["user2"] = {"name": "Bob", "email": "bob@example.com"}
    yield _mock_db
    print("Tearing down populated_db...")
    _mock_db.clear()

def test_add_user_to_empty_db(module_db_connection, empty_db):
    """Test adding a user to an initially empty database."""
    print("Running test_add_user_to_empty_db...")
    empty_db["user3"] = {"name": "Charlie", "email": "charlie@example.com"}
    assert "user3" in empty_db
    assert len(empty_db) == 1

def test_retrieve_user_from_populated_db(populated_db):
    """Test retrieving an existing user from a populated database."""
    print("Running test_retrieve_user_from_populated_db...")
    user = populated_db.get("user1")
    assert user is not None
    assert user["name"] == "Alice"
    assert user["email"] == "alice@example.com"

import pytest

def is_palindrome(s):
    """Checks if a string is a palindrome."""
    cleaned_s = "".join(filter(str.isalnum, s)).lower()
    return cleaned_s == cleaned_s[::-1]

@pytest.mark.parametrize("input_string, expected_output", [
    ("racecar", True),
    ("madam", True),
    ("A man, a plan, a canal: Panama", True), # With punctuation and spaces
    ("hello", False),
    ("Python", False),
    ("", True), # Empty string is a palindrome
    ("a", True), # Single character is a palindrome
])
def test_is_palindrome(input_string, expected_output):
    """Test the is_palindrome function with various inputs."""
    assert is_palindrome(input_string) == expected_output

# test_feature_status.py
import pytest

def divide(a, b):
    if b == 0:
        raise ZeroDivisionError("Cannot divide by zero")
    return a / b

@pytest.mark.skip(reason="This feature is not yet implemented")
def test_new_feature_logic():
    """A test for a feature that's still under development."""
    assert 1 == 2 # This test would fail, but it's skipped

@pytest.mark.skipif(
    pytest.__version__ < "8.0",
    reason="Requires pytest version 8.0 or higher"
)
def test_new_pytest_feature():
    """This test only runs if a specific Pytest version is met."""
    assert True

@pytest.mark.xfail(reason="Bug #1234: Division by zero is not handled gracefully yet")
def test_divide_by_zero_xfail():
    """This test is expected to fail due to a known bug."""
    assert divide(10, 0) == 0 # This will raise ZeroDivisionError, but it's xfailed

from typing import Optional

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from passlib.context import CryptContext

# --- Configuration ---
# You should get these from environment variables in a real application
SECRET_KEY = "your-super-secret-jwt-key" # MAKE THIS A LONG, RANDOM STRING!
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30 # Example: 30 minutes

# Password hashing context
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# OAuth2PasswordBearer to extract token from Authorization header
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") # "token" is the endpoint for login

# --- Password Hashing Functions ---
def verify_password(plain_password: str, hashed_password: str) -> bool:
    """Verifies a plain password against a hashed password."""
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    """Hashes a plain password."""
    return pwd_context.hash(password)

# --- JWT Token Functions ---
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
    """Creates a JWT access token."""
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.now(timezone.utc) + expires_delta
    else:
        expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

def decode_access_token(token: str) -> dict:
    """Decodes and validates a JWT access token."""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        # You can add more validation here, e.g., check for 'sub' or 'user_id'
        return payload
    except JWTError:
        raise credentials_exception

# --- Dependency for current user ---
async def get_current_user(token: str = Depends(oauth2_scheme)) -> dict:
    """
    Dependency to get the current user from a JWT token.
    Raises an HTTPException if the token is invalid or expired.
    """
    payload = decode_access_token(token)
    user_id: str = payload.get("sub") # 'sub' is commonly used for subject (e.g., user ID)
    if user_id is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token payload",
            headers={"WWW-Authenticate": "Bearer"},
        )
    # In a real app, you would fetch the user from your database here
    # to ensure they still exist and are active.
    # For simplicity, we'll just return the user_id for now.
    return {"user_id": user_id}

# A helper function to authenticate user (combines with your "database" logic)
def authenticate_user(username: str, password: str) -> UserInDB | None:
    user = get_user(username)
    if not user:
        return None
    if not verify_password(password, user.hashed_password):
        return None
    return user

@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    """
    Login endpoint to generate JWT access token.
    Uses OAuth2PasswordRequestForm for standard username/password input.
    """
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, # 'sub' claim typically holds the user identifier
        expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: dict = Depends(get_current_user)):
    """
    Protected endpoint: Returns information about the current authenticated user.
    Requires a valid JWT token in the Authorization header.
    """
    # In a real application, you would fetch the full user object from the DB
    # using current_user["user_id"]
    username = current_user["user_id"]
    user_data = fake_users_db.get(username)
    if not user_data:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found"
        )
    return User(**user_data)

@app.get("/protected-route/")
async def protected_route(current_user: dict = Depends(get_current_user)):
    """
    Another protected endpoint.
    """
    return {"message": f"Welcome, {current_user['user_id']}! You accessed a protected route."}

# Example of a public endpoint
@app.get("/")
async def read_root():
    return {"message": "Welcome to the unauthenticated public endpoint!"}

import os
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy.exc import SQLAlchemyError
from contextlib import contextmanager # For a clean session manager

# --- Database Configuration ---
# You'd typically get these from environment variables or a config file
DB_USER = os.environ.get("DB_USER", "myuser")
DB_PASSWORD = os.environ.get("DB_PASSWORD", "mypassword")
DB_HOST = os.environ.get("DB_HOST", "localhost")
DB_PORT = os.environ.get("DB_PORT", "3306")
DB_NAME = os.environ.get("DB_NAME", "my_test_db")

# MySQL connection string using PyMySQL driver
# Format: mysql+pymysql://user:password@host:port/dbname
DATABASE_URL = f"mysql+pymysql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"

print(f"Attempting to connect to: {DATABASE_URL}")

# --- 1. Create the Engine ---
# The engine manages the connection pool and dialect specifics.
# echo=True is great for debugging; it logs all SQL statements to console.
engine = create_engine(DATABASE_URL, echo=True, pool_pre_ping=True)

# --- 2. Define the Base ---
# Base is the declarative base class that our ORM models will inherit from.
Base = declarative_base()

# --- 3. Define the ORM Model ---
# This Python class maps to a database table.
class User(Base):
    __tablename__ = 'users' # The actual table name in the database

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, nullable=False)

    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}', email='{self.email}')>"

# --- 4. Create the Session Factory ---
# A sessionmaker factory creates Session objects.
# Sessions are the actual interface for database operations (transactions, queries).
Session = sessionmaker(bind=engine)

# --- Context Manager for Session (Best Practice) ---
# This ensures the session is properly closed even if errors occur.
@contextmanager
def get_session():
    session = Session()
    try:
        yield session
        session.commit() # Auto-commit on success
    except SQLAlchemyError as e:
        print(f"Database error occurred: {e}")
        session.rollback() # Rollback on error
        raise # Re-raise the exception after rollback
    finally:
        session.close() # Always close the session

# --- CRUD Operations ---

def create_tables():
    print("\n--- Creating tables ---")
    try:
        # Base.metadata contains all table definitions inherited from Base.
        # create_all creates these tables in the database linked by the engine.
        Base.metadata.create_all(engine)
        print("Tables created successfully.")
    except SQLAlchemyError as e:
        print(f"Error creating tables: {e}")

def create_user(name: str, email: str):
    print(f"\n--- Creating user: {name} ({email}) ---")
    with get_session() as session:
        new_user = User(name=name, email=email)
        session.add(new_user)
        print(f"Added user: {new_user}")
        return new_user

def read_users():
    print("\n--- Reading users ---")
    with get_session() as session:
        users = session.query(User).all() # Query all users
        if users:
            for user in users:
                print(user)
        else:
            print("No users found.")
        return users

def read_user_by_email(email: str):
    print(f"\n--- Reading user by email: {email} ---")
    with get_session() as session:
        user = session.query(User).filter_by(email=email).first() # Query by email
        if user:
            print(f"Found user: {user}")
        else:
            print(f"User with email '{email}' not found.")
        return user

def update_user_email(user_id: int, new_email: str):
    print(f"\n--- Updating user {user_id}'s email to {new_email} ---")
    with get_session() as session:
        user = session.query(User).filter_by(id=user_id).first()
        if user:
            old_email = user.email
            user.email = new_email
            # session.commit() is handled by the context manager on success
            print(f"Updated user {user.name} email from {old_email} to {user.email}")
            return user
        else:
            print(f"User with ID {user_id} not found.")
            return None

def delete_user(user_id: int):
    print(f"\n--- Deleting user with ID: {user_id} ---")
    with get_session() as session:
        user = session.query(User).filter_by(id=user_id).first()
        if user:
            session.delete(user)
            # session.commit() is handled by the context manager on success
            print(f"Deleted user: {user.name} ({user.id})")
            return user
        else:
            print(f"User with ID {user_id} not found.")
            return None

# --- Main Execution ---
if __name__ == "__main__":
    # Ensure environment variables are set or defaults are acceptable
    if "DB_USER" not in os.environ:
        print("WARNING: DB_USER environment variable not set. Using default 'myuser'.")
    if "DB_PASSWORD" not in os.environ:
        print("WARNING: DB_PASSWORD environment variable not set. Using default 'mypassword'.")

    # 1. Create tables (only run this once or when schema changes)
    create_tables()

    # 2. Create some users
    user1 = create_user("Alice", "alice@example.com")
    user2 = create_user("Bob", "bob@example.com")
    # Try to create user with duplicate email to see error handling
    try:
        create_user("Charlie", "alice@example.com")
    except SQLAlchemyError:
        print(" (Expected error: Duplicate email caught and rolled back)")

    # 3. Read all users
    read_users()

    # 4. Read a specific user by email
    read_user_by_email("bob@example.com")
    read_user_by_email("nonexistent@example.com")

    # 5. Update a user
    if user1: # Only if user1 was created successfully
        update_user_email(user1.id, "alice.new@example.com")
        read_user_by_email("alice.new@example.com")

    # 6. Delete a user
    if user2: # Only if user2 was created successfully
        delete_user(user2.id)
        read_users() # Show that Bob is gone

    # Try deleting a non-existent user
    delete_user(999)

    print("\n--- All operations complete ---")
    # In a real application, the engine would be disposed when the app shuts down.
    # For this script, Python will clean it up on exit.

from pathlib import Path

# Create Path Objects:
my_file = Path("data")

# Use Path methods
my_file.exists()

# Content I/O

my_file.read_text()
#Path.write_text()
#Path.write_text()
#Path.write_bytes()

# From CWD
# 
current_dir = Path.cwd()
# From Home
home_dir = Path.home()
# From absolute Paths
abs_path = Path("/usr/local/bin/python")
# From relative paths (relative to CWD)
relative_path = Path("data/input.csv")

# Create path by manipulation
base_dir = Path('/opt/my_app')
config_file = base_dire / "config" / "settings.yaml"

parent_dir = config_file.parent

# Get file / directory name
config_file.name

# Getting Stem
config_file.stem # settings

# Getting suffix
config_file.suffix
config_file.suffixes

# Get absolute path
config_file.resolve()
# or
config_file.absolute()

# Get relative path
relative_to = config_file.relative_to(project_root)

my_file.exists()

my_file.is_file()
my_file.is_dir()
my_file.is_symlink()

# Statistics
stats = temp_file.stat()

# Create directories
new_dir.mkdir()

# create empty file
empty_file.touch()

# delete file

file_to_delete.unlink()


# delete empty directories

empty_folder.rmdir()

# rename / move file or directories

old_path.rename(new_path)

# Changing suffix
config_file.with_suffix('.yml')

config_path = Path("config.txt")
config_path.write_text("debug=True\nlog_level=INFO")
content = config_path.read_text()

binary_data_file = Path("binary_data.bin")
binary_data_file.write_bytes(b'\x01\x02\x03\x04')
data = binary_data_file.read_bytes()
print(f"Binary data: {data}")

# List
project_root.iterdir()

# Globbing
project_root.glob("*.py")

# Walking Directory Tree (Python 3.12+)
project_root.walk()

try:
    with open("my_large_file.csv", "w", encoding="utf-8") as f:
        f.write("Header1,Header2\n")
        for i in range(1000):
            f.write(f"data_{i},value_{i}\n")
except IOError as e:
    print(f"Error writing file: {e}")

import tempfile
from pathlib import Path

# Using a temporary directory
with tempfile.TemporaryDirectory() as tmp_dir_str:
    tmp_dir = Path(tmp_dir_str)
    temp_file = tmp_dir / "temp_report.txt"
    temp_file.write_text("Ephemeral data.")
    print(f"Created temporary file at: {temp_file}")
    # At the end of the 'with' block, tmp_dir_str and its contents are deleted
print("Temporary directory removed.")

import shutil

source_dir = Path("my_data")
destination_dir = Path("backup_data")


try:
    shutil.copytree(source_dir, destination_dir)
    print(f"Copied '{source_dir}' to '{destination_dir}'")
except FileExistsError:
    print(f"Destination '{destination_dir}' already exists. Skipping copy.")
except Exception as e:
    print(f"Error copying tree: {e}")

import shutil
from pathlib import Path

dir_to_delete = Path("backup_data") # Assuming this exists from the copytree example

if dir_to_delete.exists():
    print(f"Deleting '{dir_to_delete}'...")
    shutil.rmtree(dir_to_delete)
    print("Directory deleted.")
else:
    print(f"Directory '{dir_to_delete}' does not exist.")

archive_path = shutil.make_archive(archive_name, 'zip', source_dir)
print(f"Created archive: {archive_path}")

usage = shutil.disk_usage(Path(".")) # Check current directory's disk
print(f"Total: {usage.total / (1024**3):.2f} GB")
print(f"Used: {usage.used / (1024**3):.2f} GB")
print(f"Free: {usage.free / (1024**3):.2f} GB")

async def my_coroutine():
    print("Coroutine started")
    await asyncio.sleep(1) # This is a pause point
    print("Coroutine resumed after 1 second")
    return "Done!"