一文でRedisの機能やユースケースをまとめてみる。

Redis とは

Redisは様々なデータ構造に対応したインメモリデータストアです。

具体的に言うと、下記のデータ構造を対応しています：

• Strings (simple binary-safe blobs)
• Lists (linked lists, with operations like LPUSH, LRANGE, etc.)
• Hashes (maps/dictionaries of field→value pairs)
• Sets (unordered collections of unique elements)
• Sorted Sets (elements ordered by a score, i.e. your “sorted array”)
• Bitmaps (compact bit arrays)
• HyperLogLogs (probabilistic cardinality counters)
• Streams (append-only log data structures)
• Geospatial indexes (store and query latitude/longitude points)

データ構造の対応ではなくて、便利な操作機能も提供しており、普段のデータベースで実装しにくい機能を簡単に実装できます。

SFTP use SSH protocal, allowing user to transfer files safely. It is a subsystem of SSH, so it runs in port 22.

The name contains FTP, however it is not implements FTP protocal

• It achieves the same function with FTP
• It is widely supported by FTP clients

When you want to transfer files with server, it might be a good choice.

How to setup a sftp user allowing it to upload files into specific folder

The following scripts helps setup a SFTP user in

Save it to .sh file, execute the following commands

chmod +x setup_sftp_user.sh
sudo ./setup_sftp_user.sh vendor

It will prompt you to set the password and configure the rest automatically.

#!/bin/bash

# Check if running as root
if [ "$EUID" -ne 0 ]; then
  echo "Please run as root or use sudo"
  exit 1
fi

# Check if username is provided
if [ -z "$1" ]; then
  echo "Usage: $0 <username>"
  exit 1
fi

USER=$1

# Create user
useradd -m $USER
passwd $USER

# Setup SFTP directories
mkdir -p /var/sftp/$USER/uploads
chown root:root /var/sftp/$USER
chmod 755 /var/sftp/$USER

chown $USER:$USER /var/sftp/$USER/uploads
chmod 755 /var/sftp/$USER/uploads

# Backup sshd_config
cp /etc/ssh/sshd_config /etc/ssh/sshd_config.bak

# Add SSH config for user
cat <<EOL >> /etc/ssh/sshd_config

Match User $USER
    ChrootDirectory /var/sftp/$USER
    ForceCommand internal-sftp
    AllowTcpForwarding no
    X11Forwarding no
EOL

# Restart SSH daemon
systemctl restart sshd

# Confirm success
echo "SFTP user '$USER' has been set up successfully."

To simplify & unify the system logic, the following convention are considered as best practice:

• Store everything in UTC
• Use UTC everywhere in backend
• Use UTC, with ISO-8601 format in API layer
• Frontend: Convert to user's local time

1. 機械学習におけるTransformer（トランスフォーマー）

現在、最も注目されているのは、深層学習モデルの一種である「Transformer」です。これは、2017年にGoogleの研究者によって発表された画期的なモデルで、特に自然言語処理（NLP）の分野に大きな変革をもたらしました。

Transformerの主な特徴とすごい点:

Attention Mechanism（注意機構）: これがTransformerの核となる技術です。従来のモデル（RNNやLSTMなど）は、文章を前から順番に処理していくため、長い文章になると前の情報を忘れやすいという問題がありました。しかし、Attention Mechanismは、入力された文章の中のどの単語とどの単語が関連しているかを計算し、「注目」することで、離れた単語間の関係性も効率的に捉えることができます。これにより、文脈をより深く理解し、より自然な言語処理が可能になりました。

並列処理の実現: 従来のRNNなどが逐次処理（単語を一つずつ処理）であったのに対し、TransformerはAttention Mechanismによって文章中の全ての単語を同時に処理する並列処理が可能です。これにより、大量のデータを高速に学習できるようになり、大規模なモデルの構築が可能になりました。

Encoder-Decoder構造: Transformerは、大きく「Encoder（エンコーダ）」と「Decoder（デコーダ）」の2つの部分で構成されています。

Encoder: 入力された文章（例えば、翻訳元の言語の文章）の重要な特徴を抽出し、その情報をベクトル表現に変換します。

Decoder: Encoderが抽出した特徴に基づいて、出力される文章（例えば、翻訳先の言語の文章）を生成します。

長期的な記憶能力: Attention Mechanismにより、長い文章でも文脈を失うことなく処理できるため、要約や文章生成など、長期的な依存関係を考慮する必要があるタスクで高い性能を発揮します。

汎用性の高さ: 元々は機械翻訳のために開発されましたが、その強力な能力から、ChatGPTやBard、Bingなどの**大規模言語モデル（LLM）**の基盤技術として広く採用されています。文章生成、要約、質問応答、感情分析など、多様な自然言語処理タスクで優れた性能を発揮します。また、画像生成や音声処理など、自然言語以外の分野にも応用されています。

「注意すべき単語」を選ぶ方法​

人間が文章を読む際に、文脈に応じて重要な単語に自然と注目するような働きを、数学的にモデル化したものです。

具体的には、以下の3つのベクトルが重要な役割を果たします。

Query（クエリ）: 今、注目している単語（例えば、ある単語を生成しようとしているとき、その単語に対応する情報）

Key（キー）: 入力された文章中の、すべての単語（または単語の表現）

Value（バリュー）: Keyに対応する情報（各単語が持つ意味的な内容）

Transformerでは、これらのベクトルを用いて、以下のような計算が行われます。

類似度の計算

まず、注目している単語のQueryベクトルと、入力文中の各単語のKeyベクトルとの「類似度」を計算します。これは一般的に、内積（ドット積）で行われます。QueryとKeyが似ていればいるほど、類似度が高くなります。

例えば、「猫が魚を食べた」という文章で、「食べた」という単語のQueryを考えます。このQueryは、入力文中の「猫」「が」「魚」「を」などのKeyとそれぞれ類似度を計算されます。

ソフトマックス関数による正規化

計算された類似度は、そのままでは数値の範囲が広すぎるため、ソフトマックス関数を通して正規化されます。これにより、類似度の合計が1になるような確率分布に変換されます。

この確率が、それぞれの単語に対する「注意の重み（Attention Weight）」となります。つまり、どの単語にどれだけ注目すべきかを示す度合いです。

先の例で言えば、「食べた」というQueryに対して、「魚」というKeyの類似度が最も高ければ、「魚」に最も大きな注意の重みが割り当てられます。

Valueの加重平均

最後に、計算された注意の重みを、各単語のValueベクトルにかけ合わせ、それらをすべて合計します。

これにより、「注意の重み付けされたValueの加重平均」が得られます。この加重平均されたベクトルが、今注目している単語にとって最も関連性の高い文脈情報を含んだものとなります。

Encoder（エンコーダ）の詳細​

Encoderは、入力系列の各単語を数値ベクトルに変換し、その単語と文脈全体との関係性を捉え、入力系列全体の意味を包括的に表現する役割を担います。

Encoderの内部構造:

Encoderは、通常、複数の identical な層（レイヤー）が積み重ねられた構造をしています。各層は主に以下の2つのサブ層から構成されます。

Multi-Head Self-Attention（マルチヘッド自己注意機構）

役割: 入力系列内の各単語が、同じ入力系列内の他のすべての単語とどれだけ関連があるかを計算し、その関連性に基づいて、各単語の表現を更新します。

Self-Attention（自己注意）: 「自己」という名前が示す通り、Query、Key、Valueの全てが、同じ入力系列の単語から導出されます。これにより、例えば「Apple（企業名）は、新しいApple（果物）を発表した」のような文で、「Apple」が文脈によって異なる意味を持つことを理解できます。

Multi-Head（マルチヘッド）: 複数の異なる「注意の視点」（Head）を持つことで、単語間の多様な関係性（例：文法的な関係、意味的な関係など）を同時に捉えることができます。それぞれのヘッドが異なるAttention Weightを計算し、それらを連結して最終的な出力とします。

出力: この層の出力は、入力単語のベクトル表現が、その単語と文脈中の他の単語との関連性によって「文脈化」されたものになります。

Feed-Forward Network（フィードフォワードネットワーク）

役割: Multi-Head Self-Attentionの出力である「文脈化された単語表現」に対して、非線形な変換を適用し、より高次元の抽象的な特徴を抽出します。

構造: 通常、活性化関数（例：ReLU）を挟んだ2層の全結合ネットワーク（Fully Connected Layer）です。各単語の表現は、このネットワークを独立して通過します。

重要な補足: Positional Encoding（位置エンコーディング）

Transformerは、RNNのように単語を順番に処理するのではなく、並列に処理します。このため、単語の位置情報が失われてしまいます。これを補うために、各単語の入力ベクトルにPositional Encodingと呼ばれる、単語の位置に応じた情報を付加します。これにより、Transformerは単語の並び順（「猫が魚を食べた」と「魚が猫を食べた」の違いなど）を認識できるようになります。

Encoderの最終的な出力:

Encoderスタックの最後の層から出力されるベクトル群は、入力文章全体の意味的・文法的な情報が凝縮された「文脈表現」となります。この情報は、Decoderに渡されます。

もちろん、TransformerのEncoder-Decoder構造について、もう少し詳しく掘り下げて説明します。

TransformerのEncoder-Decoder構造は、主にシーケンス・トゥ・シーケンス（Sequence-to-Sequence: Seq2Seq）タスク、つまり、ある系列（シーケンス）の入力を別の系列の出力に変換するタスク（例：機械翻訳、要約など）において、その威力を発揮します。

全体の流れのイメージ Encoderが「入力文の理解」を担当します。 入力された文章（例：英語の文）を読み込み、その文章のすべての情報（意味、文法構造、単語間の関係性など）を凝縮した「文脈ベクトル」のような表現（実際には複数のベクトル）を生成します。

Decoderが「出力文の生成」を担当します。 Encoderが作り出した「文脈ベクトル」と、これまでに生成した出力単語（例：日本語に翻訳された単語）に基づいて、次に続く単語を予測し、最終的な出力文章を生成します。

1. Encoder（エンコーダ）の詳細 Encoderは、入力系列の各単語を数値ベクトルに変換し、その単語と文脈全体との関係性を捉え、入力系列全体の意味を包括的に表現する役割を担います。

Encoderの内部構造:

Encoderは、通常、複数の identical な層（レイヤー）が積み重ねられた構造をしています。各層は主に以下の2つのサブ層から構成されます。

Multi-Head Self-Attention（マルチヘッド自己注意機構）

役割: 入力系列内の各単語が、同じ入力系列内の他のすべての単語とどれだけ関連があるかを計算し、その関連性に基づいて、各単語の表現を更新します。

Self-Attention（自己注意）: 「自己」という名前が示す通り、Query、Key、Valueの全てが、同じ入力系列の単語から導出されます。これにより、例えば「Apple（企業名）は、新しいApple（果物）を発表した」のような文で、「Apple」が文脈によって異なる意味を持つことを理解できます。

Multi-Head（マルチヘッド）: 複数の異なる「注意の視点」（Head）を持つことで、単語間の多様な関係性（例：文法的な関係、意味的な関係など）を同時に捉えることができます。それぞれのヘッドが異なるAttention Weightを計算し、それらを連結して最終的な出力とします。

出力: この層の出力は、入力単語のベクトル表現が、その単語と文脈中の他の単語との関連性によって「文脈化」されたものになります。

Feed-Forward Network（フィードフォワードネットワーク）

役割: Multi-Head Self-Attentionの出力である「文脈化された単語表現」に対して、非線形な変換を適用し、より高次元の抽象的な特徴を抽出します。

構造: 通常、活性化関数（例：ReLU）を挟んだ2層の全結合ネットワーク（Fully Connected Layer）です。各単語の表現は、このネットワークを独立して通過します。

重要な補足: Positional Encoding（位置エンコーディング）

Transformerは、RNNのように単語を順番に処理するのではなく、並列に処理します。このため、単語の位置情報が失われてしまいます。これを補うために、各単語の入力ベクトルにPositional Encodingと呼ばれる、単語の位置に応じた情報を付加します。これにより、Transformerは単語の並び順（「猫が魚を食べた」と「魚が猫を食べた」の違いなど）を認識できるようになります。

Encoderの最終的な出力:

Encoderスタックの最後の層から出力されるベクトル群は、入力文章全体の意味的・文法的な情報が凝縮された「文脈表現」となります。この情報は、Decoderに渡されます。

2. Decoder（デコーダ）の詳細 Decoderは、Encoderが生成した入力文脈表現と、これまでに生成した出力単語に基づいて、次の出力単語を予測し、最終的な出力系列を生成する役割を担います。

Decoderの内部構造:​

Decoderもまた、複数の identical な層が積み重ねられた構造をしています。各層は主に以下の3つのサブ層から構成されます。

Masked Multi-Head Self-Attention（マスクドマルチヘッド自己注意機構）

役割: 出力系列の生成において、未来の単語（まだ生成されていない単語）に「注意」を向けないようにするための自己注意機構です。

Masked（マスクド）: Decoderは単語を一つずつ、左から右へ（順方向へ）生成していきます。そのため、ある時点での単語を生成する際には、その時点より後の単語の情報は参照できません。Masking（マスキング）とは、この「未来の情報を見せない」ための仕組みで、Attentionの計算において、未来の単語に対応するKey-Queryの類似度をゼロ（または非常に小さな値）に設定することで、注意の重みが未来の単語に割り当てられないようにします。

出力: これまでに生成された単語の文脈を反映した表現を出力します。

Multi-Head Encoder-Decoder Attention（マルチヘッドエンコーダ-デコーダ注意機構）

役割: ここがEncoderとDecoderをつなぐ重要な部分です。Encoderが生成した入力文章の文脈情報と、Decoderがこれまでに生成した出力単語の情報を関連付けます。

Query, Key, Valueの出所:

Query: DecoderのMasked Multi-Head Self-Attention層の出力（これまでの出力単語の文脈化された表現）から生成されます。

Key: Encoderスタックの最終層の出力（入力文章の文脈表現）から生成されます。

Value: Encoderスタックの最終層の出力から生成されます。

働き: Decoderは、今生成しようとしている単語にとって、入力文章のどの部分が最も重要かを判断し、その情報に「注意」を向けます。例えば、機械翻訳で「I love sushi」を翻訳する際に、"sushi"という単語を生成する段階で、英語の「sushi」という単語に強く注意を向けるような働きをします。

出力: Encoderからの情報とDecoder自身の情報が融合された表現。

Feed-Forward Network（フィードフォワードネットワーク）

役割: Encoderと同じく、非線形な変換を適用し、より高次元の抽象的な特徴を抽出します。

Decoderの最終的な出力:

Decoderスタックの最後の層から出力されるベクトルは、最終的に線形変換層とソフトマックス関数を通じて、次に生成されるべき単語の確率分布に変換されます。最も確率の高い単語が次の出力として選択され、その単語が次のデコードステップの入力として（Positional Encodingと共に）利用されます。このプロセスが、特別な終了トークン（例： ）が生成されるまで繰り返されます。

There are many situations, that we work on one thing that not familiar with. Or there is no known best practices. But luckily we can still have framework/principles to tackle them scientifically.

To do problems solving there are 4 basic elements:

• Solving the real problem
• Build a Causal Model for knowledge
• Believe in the principle/best practice
• Get feedback and iterate the knowledge

Science

Science is a systematic enterprise that builds and organizes knowledge in the form of testable explanations and predictions about the universe.

Theory Facts

• 因果分析
• 仮説で始まる

7つの習慣の一番目の習慣は主体的であるです。

毎日起こることと反応の間で選択の自由を意識し、常に

我々の人生では毎日起こることがあります、その起こることに対して、私たちの反応があります。 その起こることと反応の間で、我々が選択する事由がある。 その選択する自由を意識し、

• pytest を使うことで、テストコードを簡潔かつ可読性の高いものにできる
• フィクスチャを活用することで、テストの前提条件の準備や後処理を柔軟に記述可能
• デコレーターによってデータ駆動テストをシンプルに記述できる
• 詳細なテスト結果を提供し、デバッグが容易
• 豊富なプラグインエコシステムにより、機能の拡張も容易

はじめに

Jest や JUnit などの他のテストフレームワークと比べて、pytest の導入は非常にシンプルです。

インストール​

pip install -U pytest

セットアップ​

pytest は、以下の簡単なルールに基づいてテストケースを自動的に検出します：

• テストファイル名：test_ で始まるか、_test.py で終わるファイル
• テスト関数/クラス：test_ で始まる関数、または Test で始まるクラスの中にある関数（ただし init は不可）

特別な設定ファイルなどは基本的に不要で、すぐにテストコードの記述を始められます。

一般的には、プロジェクトルートに tests/ ディレクトリを作成してテストコードを配置します。

my_project/
├── my_package/
│   ├── __init__.py
│   ├── module_a.py
│   └── module_b.py
├── tests/
│   ├── __init__.py           # Can be empty, but often present
│   ├── unit/                 # Optional: subdirectories for different test types
│   │   ├── test_module_a_unit.py
│   │   └── test_module_b_unit.py
│   ├── integration/          # Optional: for integration tests
│   │   └── test_api_integration.py
│   ├── functional/           # Optional: for end-to-end tests
│   │   └── test_user_flow.py
│   └── conftest.py           # For shared fixtures (more on this below)
├── README.md
└── setup.py

テストの書き方​

基本は assert を使って期待値を検証​

# 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 を使った前処理・後処理の共通化​

目的

• テスト実行前の初期状態の準備（例：DB接続、設定ファイル、モックの準備）
• テスト後のクリーンアップ処理（例：接続の切断、一時ファイル削除）
• 共通処理の再利用による重複の排除

定義方法

• 通常の Python 関数に @pytest.fixture をつけて定義します
• yield の前に前処理、後に後処理を記述します
• 使用する側では引数にフィクスチャ名を指定するだけで自動的に注入されます

利用方法

• フィクスチャを使用したいテスト関数や他のフィクスチャの引数に、定義したフィクスチャ関数名を指定するだけです。Pytestが自動的に依存性を解決し、フィクスチャが提供する値を注入してくれます。

スコープの指定

フィクスチャはデフォルトで function スコープですが、scope 引数でその実行頻度を変更できます。

• function (デフォルト): 各テスト関数ごとに1回実行。
• class: テストクラス内の全テストメソッドに対して1回実行。
• module: モジュール（ファイル）内の全テストに対して1回実行。
• session: テストセッション全体で1回だけ実行。

フィクスチャの依存関係

フィクスチャは他のフィクスチャを引数として受け取ることができます。

conftest.py を使ったフィクスチャの共有

フィクスチャは、conftest.py という特別なファイルに定義することで、複数のテストファイル間で共有することができます。

conftest.py に定義されたフィクスチャは、明示的なインポートなしに、同じディレクトリやサブディレクトリ内のすべてのテストファイルから利用可能です。これにより、プロジェクト全体で共通のセットアップロジックを一元管理できます。

@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": "[email protected]"}
    _mock_db["user2"] = {"name": "Bob", "email": "[email protected]"}
    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": "[email protected]"}
    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"] == "[email protected]"

@pytest.mark.parametrize でデータ駆動テスト​

同じロジックを異なるデータセットで繰り返しテストしたい場合、parametrize を使えばコードを繰り返す必要がありません。

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

その他​

テストをスキップする方法。 実際のプロジェクトを開発する際に、テストをスキップするのは、実務上たまに使ってるユースケースです。

Pytestを使ったら簡単にできます。

# 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

総評

pytest はシンプルながらも強力な機能を持ち、テストコードの質を大きく向上させることができます。特にフィクスチャやパラメータ化テストを活用することで、実用的で保守性の高いテストが可能です。

• Use jose to encode / decode jwt
• Use passlib to verify hashed password

Details

Utility functions​

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}

Usage

# 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!"}