IT技術で仕事を減らしたい！

ITエンジニアのメモ＋α

vsftpdでセキュアなFTPサーバを構築する！

Linux FTP

どうも、nippa です。

FTPサーバの構築は、ファイル転送を行う上で重要な技術の一つです。特にvsftpd（Very Secure FTP Daemon）は、セキュリティ面で優れた特徴を持つFTPサーバソフトウェアとして多くの環境で採用されています。本記事では、vsftpdを使用してセキュアなFTPサーバを新規構築する手順を詳しく解説します。

環境

vsftpdとは

vsftpd（Very Secure FTP Daemon）は、セキュリティを重視して設計されたFTPサーバソフトウェアです。以下の特徴があります：

主な特徴

vsftpdのインストール

パッケージのインストール

まず、vsftpdパッケージをインストールします：

# CentOS/RHEL/Rocky Linux の場合
sudo dnf install vsftpd

# Ubuntu/Debian の場合
sudo apt update
sudo apt install vsftpd

サービスの有効化

インストール後、vsftpdサービスを有効化し、起動します：

# サービスの有効化
sudo systemctl enable vsftpd

# サービスの起動
sudo systemctl start vsftpd

# サービス状態の確認
sudo systemctl status vsftpd

基本設定

設定ファイルの編集

vsftpdの主要な設定ファイルは /etc/vsftpd/vsftpd.conf です。

sudo cp /etc/vsftpd/vsftpd.conf /etc/vsftpd/vsftpd.conf.backup
sudo vi /etc/vsftpd/vsftpd.conf

基本的な設定項目

# 匿名FTPを無効化
anonymous_enable=NO

# ローカルユーザーのログインを有効化
local_enable=YES

# 書き込み権限を有効化
write_enable=YES

# umaskの設定（ファイル作成時の権限）
local_umask=022

# ローカルユーザーのchroot（ホームディレクトリに制限）
chroot_local_user=YES

# ユーザーリストの使用
userlist_enable=YES
userlist_file=/etc/vsftpd/user_list
userlist_deny=NO

# パッシブモードの設定
pasv_enable=YES
pasv_min_port=10000
pasv_max_port=10100

# ログ設定
xferlog_enable=YES
xferlog_file=/var/log/xferlog
log_ftp_protocol=YES

# IPv4の使用を強制
listen=YES
listen_ipv6=NO

ユーザー管理

FTPユーザーの作成

専用のFTPユーザーを作成します：

# FTPユーザーの作成
sudo useradd -m -s /bin/bash ftpuser

# パスワードの設定
sudo passwd ftpuser

# FTPホームディレクトリの作成
sudo mkdir -p /home/ftpuser/ftp
sudo chown nobody:nogroup /home/ftpuser/ftp
sudo chmod a-w /home/ftpuser/ftp

# アップロード用ディレクトリの作成
sudo mkdir /home/ftpuser/ftp/upload
sudo chown ftpuser:ftpuser /home/ftpuser/ftp/upload

ユーザーリストの設定

FTPアクセスを許可するユーザーを /etc/vsftpd/user_list に追加：

sudo echo "ftpuser" >> /etc/vsftpd/user_list

SSL/TLS暗号化の設定

SSL証明書の生成

自己署名証明書を作成します：

sudo openssl req -x509 -nodes -days 365 -newkey rsa:2048 \
    -keyout /etc/vsftpd/private/vsftpd.key \
    -out /etc/vsftpd/private/vsftpd.pem \
    -subj "/C=JP/ST=Tokyo/L=Tokyo/O=YourOrg/CN=your-server.com"

sudo chmod 600 /etc/vsftpd/private/vsftpd.*

SSL設定の追加

vsftpd.confにSSL設定を追加：

# SSL/TLS暗号化を有効化
ssl_enable=YES
ssl_tlsv1=YES
ssl_sslv2=NO
ssl_sslv3=NO

# 証明書とキーファイルの指定
rsa_cert_file=/etc/vsftpd/private/vsftpd.pem
rsa_private_key_file=/etc/vsftpd/private/vsftpd.key

# SSL接続の強制
force_local_data_ssl=YES
force_local_logins_ssl=YES

# SSL設定の詳細
ssl_ciphers=HIGH
require_ssl_reuse=NO

ファイアウォール設定

ポートの開放

FTPサーバに必要なポートを開放します：

# FTPコントロールポート（21番）
sudo firewall-cmd --permanent --add-service=ftp

# パッシブモード用ポート範囲
sudo firewall-cmd --permanent --add-port=10000-10100/tcp

# 設定の反映
sudo firewall-cmd --reload

SELinuxの設定

SELinuxが有効な場合は、FTP用の設定を行います：

# FTPのホームディレクトリアクセスを許可
sudo setsebool -P ftp_home_dir on

# FTPユーザーの書き込みを許可
sudo setsebool -P allow_ftpd_full_access on

サービスの再起動と動作確認

設定の反映

設定を変更したら、vsftpdサービスを再起動します：

# 設定ファイルの構文チェック
sudo vsftpd -olisten=NO /etc/vsftpd/vsftpd.conf

# サービスの再起動
sudo systemctl restart vsftpd

# サービス状態の確認
sudo systemctl status vsftpd

接続テスト

FTPクライアントで接続をテストします：

# コマンドラインでのテスト
ftp your-server-ip

# または、SFTPクライアントでの接続確認
sftp ftpuser@your-server-ip

ログ監視とメンテナンス

ログファイルの確認

FTPサーバのログを定期的に確認します：

# 転送ログの確認
sudo tail -f /var/log/xferlog

# vsftpdログの確認（systemdの場合）
sudo journalctl -u vsftpd -f

# 認証ログの確認
sudo tail -f /var/log/secure

定期メンテナンス

# ログファイルのローテーション設定
sudo vi /etc/logrotate.d/vsftpd

# 不要なログファイルの削除
sudo find /var/log -name "*.log" -mtime +30 -delete

セキュリティ強化

追加のセキュリティ設定

より安全なFTPサーバにするための追加設定：

# 接続試行回数の制限
max_login_fails=3

# 接続タイムアウトの設定
idle_session_timeout=300
data_connection_timeout=120

# 匿名アップロードの禁止
anon_upload_enable=NO
anon_mkdir_write_enable=NO

# ローカルユーザーの一覧表示を禁止
hide_ids=YES

# banner設定
ftpd_banner=Welcome to FTP Server

fail2banの設置

不正アクセスを防ぐためfail2banを設置：

# fail2banのインストール
sudo dnf install epel-release
sudo dnf install fail2ban

# vsftpd用設定の作成
sudo vi /etc/fail2ban/jail.local

トラブルシューティング

よくある問題と解決方法

問題1: 500 OOPS: vsftpd: refusing to run with writable root inside chroot()

# 解決方法: FTPルートディレクトリの書き込み権限を削除
sudo chmod a-w /home/ftpuser/ftp

問題2: パッシブモードで接続できない

# パッシブモードの設定を確認・修正
pasv_enable=YES
pasv_min_port=10000
pasv_max_port=10100
pasv_address=your-public-ip

問題3: SSL接続でエラーが発生

# SSL証明書の権限を確認
sudo chmod 600 /etc/vsftpd/private/vsftpd.*
sudo chown root:root /etc/vsftpd/private/vsftpd.*

感想

vsftpdを使用したFTPサーバの構築は、適切な設定を行うことで高いセキュリティと安定性を実現できます。特にSSL/TLS暗号化の設定により、データ転送の安全性を大幅に向上させることができます。定期的なログ監視とメンテナンスを行い、セキュアなファイル転送環境を維持していきましょう。

ではでは、また次回。

Rust環境構築セットアップ

Rust

どうも、nippa です。

Rustはシステムプログラミングに最適な言語として人気が急上昇しています。メモリ安全性とパフォーマンスを両立し、WebAssemblyやBlockchain開発でも注目されています。本記事では2025年時点での最新Rust環境構築方法を解説します。

環境

  • macOS 15.5 / Windows 11 / Ubuntu 22.04+
  • Rust 1.75+
  • Cargo（Rustのパッケージマネージャー）

Rustupによるインストール

macOS / Linux

# Rustupインストーラーをダウンロード・実行
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh

# 環境変数を現在のシェルに反映
source ~/.cargo/env

# インストール確認
rustc --version
cargo --version

Windows

PowerShellで実行：

# Rustupインストーラーをダウンロード
Invoke-WebRequest -Uri "https://win.rustup.rs/" -OutFile "rustup-init.exe"

# インストーラー実行
.\rustup-init.exe

# パス設定（再起動後に有効）
$env:Path += ";$env:USERPROFILE\.cargo\bin"

または、公式サイトからrustup-init.exeをダウンロードして実行。

パッケージマネージャーによるインストール

Homebrew（macOS）

# Rustインストール
brew install rust

# 追加ツールのインストール
brew install rustup-init
rustup-init

apt（Ubuntu/Debian）

# システムのRust (古いバージョンの場合)
sudo apt update
sudo apt install rustc cargo

# 最新版はRustupを推奨
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh

Chocolatey（Windows）

# Chocolateyでインストール
choco install rust

# またはRustupを使用
choco install rustup.install
rustup toolchain install stable

基本設定とツールチェーン管理

ツールチェーンの管理

# 利用可能なツールチェーン確認
rustup toolchain list

# 安定版を既定に設定
rustup default stable

# 最新版に更新
rustup update

# 特定バージョンのインストール
rustup toolchain install 1.75.0
rustup default 1.75.0

ターゲットの追加

# WebAssemblyターゲット追加
rustup target add wasm32-unknown-unknown

# Apple Silicon用
rustup target add aarch64-apple-darwin

# Windows用（クロスコンパイル）
rustup target add x86_64-pc-windows-gnu

# ターゲット一覧確認
rustup target list --installed

開発ツールのセットアップ

必須開発ツール

# Cargoの拡張ツール
cargo install cargo-edit        # 依存関係管理
cargo install cargo-watch       # ファイル監視・自動ビルド
cargo install cargo-expand      # マクロ展開表示
cargo install cargo-outdated    # 依存関係の更新確認

# フォーマッター・リンター
rustup component add rustfmt    # コードフォーマッター
rustup component add clippy     # リンター

# デバッグツール
cargo install cargo-edit
cargo install flamegraph        # プロファイリング

LSP（Language Server Protocol）設定

# rust-analyzerインストール
rustup component add rust-analyzer

# 確認
rust-analyzer --version

エディタ・IDE設定

VS Code

拡張機能をインストール：

{
  "recommendations": [
    "rust-lang.rust-analyzer",
    "vadimcn.vscode-lldb",
    "serayuzgur.crates"
  ]
}

settings.json設定：

{
  "rust-analyzer.check.command": "clippy",
  "rust-analyzer.cargo.features": "all",
  "rust-analyzer.inlayHints.enable": true,
  "[rust]": {
    "editor.defaultFormatter": "rust-lang.rust-analyzer",
    "editor.formatOnSave": true
  }
}

Vim/Neovim

-- init.lua (Neovim)
require('mason').setup()
require('mason-lspconfig').setup({
  ensure_installed = { "rust_analyzer" }
})

local lspconfig = require('lspconfig')
lspconfig.rust_analyzer.setup({
  settings = {
    ["rust-analyzer"] = {
      checkOnSave = {
        command = "clippy"
      }
    }
  }
})

JetBrains IntelliJ IDEA

  1. Rustプラグインをインストール
  2. File → Settings → Languages & Frameworks → Rust
  3. Toolchain path: ~/.cargo/bin
  4. Standard library: 自動検出

最初のプロジェクト作成

プロジェクト初期化

# 新規プロジェクト作成
cargo new hello_rust
cd hello_rust

# または既存ディレクトリを初期化
cargo init

プロジェクト構造

hello_rust/
├── Cargo.toml      # 設定・依存関係
├── src/
│   └── main.rs     # メインソースファイル
└── target/         # ビルド出力（自動生成）

基本的なCargo.toml

[package]
name = "hello_rust"
version = "0.1.0"
edition = "2021"

[dependencies]
serde = { version = "1.0", features = ["derive"] }
tokio = { version = "1.0", features = ["full"] }
clap = { version = "4.0", features = ["derive"] }

[dev-dependencies]
proptest = "1.0"

実行とテスト

ビルド・実行

# ビルド
cargo build

# リリースビルド
cargo build --release

# 実行
cargo run

# 引数付き実行
cargo run -- --help

# 特定のバイナリ実行
cargo run --bin my_binary

テスト

# テスト実行
cargo test

# 特定のテスト実行
cargo test test_name

# テストを詳細表示
cargo test -- --nocapture

# ベンチマークテスト
cargo test --release

フォーマット・リント

# コードフォーマット
cargo fmt

# リント実行
cargo clippy

# リント（警告をエラーとして扱う）
cargo clippy -- -D warnings

便利なCargo設定

~/.cargo/config.toml

[build]
# デフォルトターゲット
target = "x86_64-apple-darwin"

[target.x86_64-apple-darwin]
# リンカー設定
linker = "clang"
rustflags = ["-C", "link-arg=-fuse-ld=lld"]

[registries.crates-io]
protocol = "sparse"

[net]
# プロキシ設定（必要に応じて）
# git-fetch-with-cli = true

[alias]
# カスタムエイリアス
b = "build"
r = "run"
t = "test"
c = "clippy"

プロジェクト固有設定

# .cargo/config.toml (プロジェクトルート)
[env]
RUST_LOG = "debug"
DATABASE_URL = "sqlite://./app.db"

[build]
rustflags = ["-C", "target-cpu=native"]

トラブルシューティング

よくある問題と解決策

1. リンカーエラー

# macOSの場合
xcode-select --install

# Ubuntuの場合
sudo apt install build-essential

# Windows（Visual Studio Build Tools必要）
# Visual Studio Installerで「C++ build tools」をインストール

2. 権限エラー

# Cargoディレクトリの権限修正
chmod -R 755 ~/.cargo

3. プロキシ環境での問題

# 環境変数設定
export HTTPS_PROXY=http://proxy.company.com:8080
export HTTP_PROXY=http://proxy.company.com:8080

# Git設定
git config --global http.proxy http://proxy.company.com:8080

4. 古いツールチェーンの問題

# 完全クリーンアップ
rustup self uninstall
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh

次のステップ

環境構築完了後の学習リソース：

  1. 公式ドキュメント: The Rust Programming Language
  2. Rustlings: 実践的な練習問題
  3. Cargo Book: パッケージ管理の詳細
  4. Rust by Example: コード例で学習
# Rustlings（練習問題）のインストール
cargo install rustlings
rustlings init
rustlings watch

感想

Rust環境構築はRustupを使用することで非常にシンプルになりました。ツールチェーン管理、クロスコンパイル、エディタ連携も充実しており、モダンな開発環境を簡単に構築できます。メモリ安全性とパフォーマンスを両立するRustで、システムプログラミングの新しい世界を探索してみてください。

ではでは、また次回。

Microsoft MarkItDown: 文書デジタル化とAI活用の実用的ツール

Python

どうも、nippa です。

Microsoftがリリースした「MarkItDown」は、PDF、Word、Excel、PowerPointなど様々なファイル形式をMarkdownに変換できるPythonツールです。LLMでの文書処理やテキスト解析パイプラインに最適化されており、文書構造を保持しながら効率的に変換できます。

環境

MarkItDownとは

MarkItDownは、様々なファイル形式をMarkdownに変換するMicrosoft製のオープンソースツールです。従来のtextractと比較して、文書構造（見出し、リスト、表、リンクなど）を保持しながらMarkdownに変換することに特化しています。

対応ファイル形式

インストール

全機能インストール

poetry add 'markitdown[all]'

選択的インストール

# PDF、Word、PowerPointのみ
poetry add 'markitdown[pdf,docx,pptx]'

# Excel関連のみ
poetry add 'markitdown[xlsx,xls]'

# 音声・動画関連
poetry add 'markitdown[audio-transcription,youtube-transcription]'

基本的な使用方法

コマンドライン

# ファイルをMarkdownに変換
markitdown document.pdf > output.md

# 出力ファイル指定
markitdown presentation.pptx -o output.md

# パイプ処理
cat document.pdf | markitdown > output.md

Python API

from markitdown import MarkItDown

# 基本的な変換
md = MarkItDown()
result = md.convert("document.pdf")
print(result.text_content)

# ファイル情報も取得
print(f"タイトル: {result.title}")
print(f"メタデータ: {result.metadata}")

実践的な使用例

LLMとの連携

from markitdown import MarkItDown
from openai import OpenAI

# OpenAI APIクライアント設定
client = OpenAI()
md = MarkItDown(llm_client=client, llm_model="gpt-4o")

# 画像ファイルの処理（LLMによる画像説明を含む）
result = md.convert("chart.png")
print(result.text_content)

一括変換スクリプト

import os
from pathlib import Path
from markitdown import MarkItDown

def batch_convert(input_dir, output_dir):
    md = MarkItDown()
    input_path = Path(input_dir)
    output_path = Path(output_dir)
    output_path.mkdir(exist_ok=True)

    # 対応ファイル形式
    supported_extensions = ['.pdf', '.docx', '.pptx', '.xlsx', '.html']

    for file_path in input_path.iterdir():
        if file_path.suffix.lower() in supported_extensions:
            try:
                print(f"変換中: {file_path.name}")
                result = md.convert(str(file_path))

                # 出力ファイル名を生成
                output_file = output_path / f"{file_path.stem}.md"

                with open(output_file, 'w', encoding='utf-8') as f:
                    f.write(result.text_content)

                print(f"完了: {output_file}")

            except Exception as e:
                print(f"エラー ({file_path.name}): {e}")

# 実行例
batch_convert("./documents", "./markdown_output")

Azure Document Intelligenceとの連携

from markitdown import MarkItDown

# Azure Document Intelligence使用
md = MarkItDown(docintel_endpoint="<your_endpoint>")
result = md.convert("complex_document.pdf")
print(result.text_content)

YouTube動画の転写

from markitdown import MarkItDown

md = MarkItDown()
# YouTube URLから転写テキストを取得
result = md.convert("https://www.youtube.com/watch?v=example")
print(result.text_content)

Docker使用

# Dockerfile作成
FROM python:3.11-slim

RUN pip install 'markitdown[all]'

WORKDIR /app
COPY . .

ENTRYPOINT ["markitdown"]

# ビルドと実行
docker build -t markitdown:latest .
docker run --rm -i markitdown:latest < document.pdf > output.md

プラグインシステム

利用可能なプラグイン確認

markitdown --list-plugins

プラグインを有効にして変換

markitdown --use-plugins document.pdf -o output.md

カスタムプラグイン開発

from markitdown.plugin import MarkItDownPlugin

class CustomPlugin(MarkItDownPlugin):
    def can_convert(self, file_path):
        return file_path.endswith('.custom')

    def convert(self, file_path):
        # カスタム変換ロジック
        with open(file_path, 'r') as f:
            content = f.read()
        return f"# Custom File\n\n{content}"

実際の変換例

Excel → Markdown

# input.xlsx
md = MarkItDown()
result = md.convert("sales_data.xlsx")
print(result.text_content)

出力例：

# Sheet1

| 月 | 売上 | 前年比 |
|---|---|---|
| 1月 | 1,000,000 | 105% |
| 2月 | 1,200,000 | 110% |
| 3月 | 1,100,000 | 102% |

PowerPoint → Markdown

result = md.convert("presentation.pptx")

出力例：

# 企業戦略プレゼンテーション

## スライド 1: 概要
- 売上向上施策
- コスト削減計画
- 新規事業展開

## スライド 2: 詳細データ
[グラフの説明テキスト]

他ツールとの比較

機能 MarkItDown textract pandoc
Markdown特化 ✅ ❌ ✅
構造保持 ✅ 限定的 ✅
LLM連携 ✅ ❌ ❌
音声転写 ✅ ❌ ❌
YouTube対応 ✅ ❌ ❌
プラグイン ✅ ❌ ✅

感想

Microsoft MarkItDownは、文書のデジタル化とLLMでの活用を念頭に置いた実用的なツールです。特に企業環境でのOffice文書処理や、AIを活用した文書解析パイプラインの構築において威力を発揮します。プラグインシステムによる拡張性も魅力的です。

ではでは、また次回。

Python httpxで爆速非同期HTTP通信を実装する

Python

どうも、nippa です。

PythonでHTTP通信を行う際、従来のrequestsライブラリは同期処理のため大量のAPIリクエストで性能ボトルネックとなります。httpxライブラリを使用することで、非同期処理による高速なHTTP通信が実現できます。

環境

httpxとrequestsの違い

requests（同期処理）

import requests
import time

def fetch_sync(urls):
    results = []
    start_time = time.time()

    for url in urls:
        response = requests.get(url)
        results.append(response.json())

    print(f"同期処理時間: {time.time() - start_time:.2f}秒")
    return results

httpx（非同期処理）

import httpx
import asyncio
import time

async def fetch_async(urls):
    results = []
    start_time = time.time()

    async with httpx.AsyncClient() as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)

        for response in responses:
            results.append(response.json())

    print(f"非同期処理時間: {time.time() - start_time:.2f}秒")
    return results

基本的な使用方法

インストール

pip install httpx

同期処理での使用

import httpx

# GETリクエスト
response = httpx.get('https://api.github.com/users/octocat')
print(response.json())

# POSTリクエスト
data = {'key': 'value'}
response = httpx.post('https://httpbin.org/post', json=data)
print(response.status_code)

# ヘッダー付きリクエスト
headers = {'Authorization': 'Bearer token'}
response = httpx.get('https://api.example.com/data', headers=headers)

非同期処理での使用

import httpx
import asyncio

async def main():
    async with httpx.AsyncClient() as client:
        # 単一リクエスト
        response = await client.get('https://api.github.com/users/octocat')
        print(response.json())

        # 複数の並行リクエスト
        urls = [
            'https://api.github.com/users/octocat',
            'https://api.github.com/users/defunkt',
            'https://api.github.com/users/pjhyett'
        ]

        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)

        for response in responses:
            print(f"Status: {response.status_code}, User: {response.json()['login']}")

# 実行
asyncio.run(main())

実践的な使用例

API レート制限を考慮した実装

import httpx
import asyncio
from asyncio import Semaphore

class RateLimitedClient:
    def __init__(self, rate_limit=10):
        self.semaphore = Semaphore(rate_limit)
        self.client = httpx.AsyncClient(
            timeout=httpx.Timeout(10.0),
            limits=httpx.Limits(max_keepalive_connections=20, max_connections=100)
        )

    async def get(self, url):
        async with self.semaphore:
            try:
                response = await self.client.get(url)
                response.raise_for_status()
                return response
            except httpx.HTTPError as e:
                print(f"HTTP error occurred: {e}")
                return None

    async def close(self):
        await self.client.aclose()

async def fetch_with_rate_limit():
    client = RateLimitedClient(rate_limit=5)

    urls = [f'https://jsonplaceholder.typicode.com/posts/{i}' for i in range(1, 21)]

    tasks = [client.get(url) for url in urls]
    responses = await asyncio.gather(*tasks, return_exceptions=True)

    valid_responses = [r for r in responses if r and not isinstance(r, Exception)]
    print(f"成功: {len(valid_responses)}/{len(urls)} リクエスト")

    await client.close()

asyncio.run(fetch_with_rate_limit())

エラーハンドリングとリトライ機能

import httpx
import asyncio
from typing import List, Optional

async def fetch_with_retry(
    client: httpx.AsyncClient,
    url: str,
    max_retries: int = 3
) -> Optional[httpx.Response]:

    for attempt in range(max_retries):
        try:
            response = await client.get(url)
            response.raise_for_status()
            return response

        except httpx.TimeoutException:
            print(f"Timeout for {url}, attempt {attempt + 1}")
            if attempt == max_retries - 1:
                return None
            await asyncio.sleep(2 ** attempt)  # 指数バックオフ

        except httpx.HTTPStatusError as e:
            if e.response.status_code >= 500:
                print(f"Server error {e.response.status_code} for {url}")
                if attempt == max_retries - 1:
                    return None
                await asyncio.sleep(2 ** attempt)
            else:
                print(f"Client error {e.response.status_code} for {url}")
                return None

        except Exception as e:
            print(f"Unexpected error for {url}: {e}")
            return None

    return None

async def batch_fetch_with_retry():
    urls = [
        'https://jsonplaceholder.typicode.com/posts/1',
        'https://jsonplaceholder.typicode.com/posts/2',
        'https://httpbin.org/delay/2',  # 遅いエンドポイント
        'https://httpbin.org/status/500',  # エラーエンドポイント
    ]

    async with httpx.AsyncClient(timeout=5.0) as client:
        tasks = [fetch_with_retry(client, url) for url in urls]
        responses = await asyncio.gather(*tasks)

        successful = [r for r in responses if r is not None]
        print(f"成功したリクエスト: {len(successful)}/{len(urls)}")

        for response in successful:
            print(f"Status: {response.status_code}, URL: {response.url}")

asyncio.run(batch_fetch_with_retry())

ストリーミング処理

import httpx
import asyncio

async def download_large_file():
    url = 'https://httpbin.org/stream/1000'

    async with httpx.AsyncClient() as client:
        async with client.stream('GET', url) as response:
            print(f"Content-Length: {response.headers.get('content-length')}")

            total_size = 0
            async for chunk in response.aiter_bytes(chunk_size=8192):
                total_size += len(chunk)
                # ここでチャンクを処理
                print(f"Downloaded: {total_size} bytes", end='\r')

            print(f"\nTotal downloaded: {total_size} bytes")

asyncio.run(download_large_file())

パフォーマンス比較

100個のAPIリクエストの処理時間

import time
import asyncio
import httpx
import requests

def benchmark_sync():
    urls = [f'https://jsonplaceholder.typicode.com/posts/{i}'
            for i in range(1, 101)]

    start_time = time.time()
    for url in urls:
        response = requests.get(url)
    sync_time = time.time() - start_time
    return sync_time

async def benchmark_async():
    urls = [f'https://jsonplaceholder.typicode.com/posts/{i}'
            for i in range(1, 101)]

    start_time = time.time()
    async with httpx.AsyncClient() as client:
        tasks = [client.get(url) for url in urls]
        await asyncio.gather(*tasks)
    async_time = time.time() - start_time
    return async_time

# ベンチマーク実行
sync_time = benchmark_sync()
async_time = asyncio.run(benchmark_async())

print(f"同期処理: {sync_time:.2f}秒")
print(f"非同期処理: {async_time:.2f}秒")
print(f"速度向上: {sync_time/async_time:.1f}倍")

ベストプラクティス

1. AsyncClientの適切な管理

# 推奨: コンテキストマネージャーを使用
async with httpx.AsyncClient() as client:
    response = await client.get(url)

# 非推奨: 手動でのクローズ管理
client = httpx.AsyncClient()
try:
    response = await client.get(url)
finally:
    await client.aclose()

2. 適切なタイムアウト設定

timeout = httpx.Timeout(
    connect=5.0,    # 接続タイムアウト
    read=10.0,      # 読み取りタイムアウト
    write=5.0,      # 書き込みタイムアウト
    pool=10.0       # プールタイムアウト
)

async with httpx.AsyncClient(timeout=timeout) as client:
    response = await client.get(url)

3. コネクションプールの最適化

limits = httpx.Limits(
    max_keepalive_connections=20,
    max_connections=100,
    keepalive_expiry=30.0
)

async with httpx.AsyncClient(limits=limits) as client:
    # リクエスト処理
    pass

感想

httpxライブラリを使った非同期HTTP処理により、従来のrequestsと比べて大幅な性能向上が期待できます。特に複数のAPIを並行して処理する場面では、その効果は絶大です。適切なエラーハンドリングとレート制限を組み合わせることで、実用的な高性能HTTPクライアントを構築できます。

ではでは、また次回。

Poetry v2でのrequirements.txt出力方法

Python

どうも、nippa です。

Poetry 2.0からrequirements.txtの出力方法が大きく変わりました。従来のpoetry exportコマンドが廃止され、新しいpoetry exportプラグインまたはpoetry requirementsコマンドを使用します。

環境

  • Poetry 2.0以降
  • Python 3.11

Poetry 1.x系からの変更点

Poetry 1.x系（旧バージョン）

# 従来の方法（Poetry 2.0では利用不可）
poetry export -f requirements.txt --output requirements.txt

Poetry 2.x系（現在）

# 新しい方法
poetry requirements --output requirements.txt

基本的な使用方法

標準出力

# requirements.txtの内容を標準出力
poetry requirements

出力例：

certifi==2023.7.22
charset-normalizer==3.3.2
idna==3.4
requests==2.31.0
urllib3==2.0.7

ファイル出力

# requirements.txtファイルに出力
poetry requirements --output requirements.txt

# 本番環境用（dev依存関係を除外）
poetry requirements --only main --output requirements.txt

# 開発環境用のみ
poetry requirements --only dev --output requirements-dev.txt

主要オプション

依存関係のフィルタリング

# メイン依存関係のみ
poetry requirements --only main

# 開発依存関係のみ
poetry requirements --only dev

# 特定のグループのみ
poetry requirements --only test

# 複数グループ指定
poetry requirements --only main,test

形式指定

# デフォルト形式
poetry requirements

# ハッシュ付きで出力
poetry requirements --hash

# extras指定
poetry requirements --extras "dev,test"

その他のオプション

# バージョン制約なし（最新バージョン指定）
poetry requirements --without-hashes

# 出力先ディレクトリ指定
poetry requirements --output ./docker/requirements.txt

実践的な使用例

Docker環境での活用

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# Poetry設定ファイルをコピー
COPY pyproject.toml poetry.lock ./

# Poetry をインストール
RUN pip install poetry

# requirements.txtを生成
RUN poetry requirements --output requirements.txt

# 依存関係をインストール
RUN pip install -r requirements.txt

COPY . .

CMD ["python", "app.py"]

CI/CDでの使用

# GitHub Actions
name: Export Requirements
on: [push]

jobs:
  export:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Install Poetry
        run: |
          curl -sSL https://install.python-poetry.org | python3 -
          echo "$HOME/.local/bin" >> $GITHUB_PATH

      - name: Export requirements
        run: |
          poetry requirements --output requirements.txt
          poetry requirements --only dev --output requirements-dev.txt

      - name: Upload requirements
        uses: actions/upload-artifact@v3
        with:
          name: requirements
          path: requirements*.txt

Makefileでの自動化

.PHONY: requirements
requirements:
    poetry requirements --output requirements.txt
    poetry requirements --only dev --output requirements-dev.txt
    @echo "Requirements files generated successfully"

.PHONY: docker-build
docker-build: requirements
    docker build -t myapp .

Poetry Exportプラグインとの比較

プラグイン版（poetry-plugin-export）

# プラグインインストール
poetry self add poetry-plugin-export

# 使用方法（従来と同様）
poetry export -f requirements.txt --output requirements.txt

標準版との違い

機能 poetry requirements poetry export（プラグイン）
インストール 標準で利用可能 プラグイン要
機能 基本的なexport 高機能なexport
互換性 Poetry 2.0+ Poetry 1.x系と互換

トラブルシューティング

コマンドが見つからない場合

# Poetry バージョン確認
poetry --version

# Poetry 2.0未満の場合はアップグレード
curl -sSL https://install.python-poetry.org | python3 -

依存関係の解決エラー

# ロックファイルを再生成
poetry lock --no-update

# キャッシュクリア
poetry cache clear pypi --all

感想

Poetry 2系でのpoetry requirementsコマンドは、従来のexport機能をより直感的にしたものです。Docker環境やCI/CDパイプラインでの活用により、Poetryプロジェクトと従来のPythonエコシステムの橋渡しが簡単になりました。

ではでは、また次回。