pythonコマンドでこれ実行
python
from app import db
db.create_all()
デバッグモードで起動
email = StringField('Email', validators=[Optional(), Email()])
ログインしたら、mypage表示 処理
@app.route("/mypage")
@login_required
def mypage():
return render_template('mypage.html', username=current_user.username)
ログインできたら、ログイン成功画面表示
@app.route("/register", methods=["GET", "POST"])
def register():
db.session.add(new_user)
db.session.commit()
login_user(new_user)
flash('登録が完了!ログインしてね。', 'success')
return redirect(url_for('register_complete'))
サインアップ完了画面とapp.py
app.py
@app.route('/register_complete')
@login_required
def register_complete():
return render_template('register_complete.html')
html
<!doctype html>
<html lang="ja">
<head>
<meta charset="UTF-8">
<title>登録完了</title>
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css" rel="stylesheet">
</head>
<body class="bg-light">
<div class="container mt-5">
<h2 class="mb-4">登録完了!</h2>
<p>登録が完了しました。ログインしてください。</p>
<a href="{{ url_for('login') }}" class="btn btn-success">ログイン画面へ</a>
</div>
</body>
</html>
パスワードリセット / メール認証準備
pip install itsdangerous
ログイン状態を保持するボタン
app.py
from wtforms import BooleanField
remember = BooleanField('ログイン状態を保持する')
ログインボタンと登録ボタン
from wtforms import SubmitField
submit = SubmitField('登録')
submit = SubmitField('ログイン')
[空欄時のエラー]パスワード/ ユーザー入力欄が空欄時
from wtforms import StringField, PasswordField
from wtforms.validators import InputRequired
class LoginForm(FlaskForm):
username = StringField('ユーザー名', validators=[
InputRequired(message="ユーザー名は必須です。")
])
password = PasswordField('パスワード', validators=[
InputRequired(message="パスワードは必須です。")
])
[新規登録時: 既に登録済み]ユーザー / メールアドレスが登録されてたらエラー出す
新規登録モデル内で実行(RegisterForm)
from wtforms.validators import Email, ValidationError
class RegisterForm(FlaskForm):
※ 処理
※ ↓ 処理と同じインデントで書く
def validate_username(self, username):
user = User.query.filter_by(username=username.data).first()
if user:
raise ValidationError('このユーザー名は既に使用されています。')
def validate_email(self, email):
user = User.query.filter_by(email=email.data).first()
if user:
raise ValidationError('このメールアドレスは既に登録されています。')
パスワ / ユーザー名を何文字以上じゃないと登録しない
from wtforms import StringField, PasswordField
from wtforms.validators import Length
class RegisterForm(FlaskForm):
username = StringField('ユーザー名', validators=[Length(min=4, max=150, message="ユーザー名は4文字以上150文字以下にしてください。")])
password = PasswordField('パスワード', validators=[Length(min=4, max=150, message="パスワードは4文字以上150文字以下にしてください。")])
[エラーメッセージ出したい] ログイン時/ 新規登録時
ログイン時 : LoginForm(FlaskForm)モデル
新規登録時 : RegisterForm(FlaskForm):モデル
に書く
パスワリセットとメール認証 仕様
【パスワードリセット】
メアド入力 → リセットリンク送信
リンクから新しいパスワード設定
【メール認証】
新規登録時に、本人のメアドへ確認リンク送信
リンクをクリックしないとログインできない
[メアドエラー追加]Userモデルに「email」「email_verified」追加
email = db.Column(db.String(150), nullable=False, unique=True)
email_verified = db.Column(db.Boolean, default=False)
email / emai_verified 意味
email : メール入力欄
emai_verified : メール認証できてるかできてないかチェックするやつ (普段は、false)
[メアドエラー追加]RegisterFormモデルに空欄時と既に登録済み時のエラーメッセージ追加
class RegisterForm(FlaskForm):
email = StringField('メールアドレス', validators=[
InputRequired(message="メールアドレスは必須です。"),
Email(message="有効なメールアドレスを入力してください。")
])
def validate_email(self, email):
user = User.query.filter_by(email=email.data).first()
if user:
raise ValidationError('このメールアドレスは既に登録されています。')
Eメールが空欄と空欄じゃない場合に登録したい
email = StringField('Email', validators=[Optional(), Email()])
[メール認証] 新規登録処理で、メール認証で登録とメール認証なしで登録、に分ける
form.email.data:で処理分ける
if form.validate_on_submit():
hashed_pw = bcrypt.generate_password_hash(form.password.data).decode('utf-8')
# メールアドレスが入力されていたら「メール認証あり」
if form.email.data:
user = User(username=form.username.data, email=form.email.data, password=hashed_pw)
db.session.add(user)
db.session.commit()
token = s.dumps(user.email, salt='email-confirm')
link = url_for('confirm_email', token=token, _external=True)
msg = Message('メール認証を完了してください', sender='あなたのメールアドレス', recipients=[user.email])
msg.body = f'こちらのリンクをクリックして認証を完了してください: {link}'
mail.send(msg)
flash('認証メールを送りました。メールを確認してください!', 'success')
# メールアドレスなし → 通常登録
else:
new_user = User(username=form.username.data, password=hashed_pw)
db.session.add(new_user)
db.session.commit()
flash('登録が完了しました!ログインしてください。', 'success')
return redirect(url_for('register_complete'))
[メール認証]トークン有効期限設定追加 (パスワード設定リンク)/ 期限が切れたらエラーメッセージ表示
from flask_mail import Mail, Message
from wtforms.validators import Email, EqualTo
from itsdangerous import URLSafeTimedSerializer, SignatureExpired
@app.route('/confirm_email/<token>')
def confirm_email(token):
try:
email = s.loads(token, salt='email-confirm', max_age=3600) # 1時間有効
except SignatureExpired:
return '<h1>リンクの有効期限が切れています。</h1>'
user = User.query.filter_by(email=email).first_or_404()
メール認証済みならエラー表示し、メール認証されたら"メール認証できた"と表示し、ログイン画面へ行かせる
@app.route('/confirm_email/<token>')
def confirm_email(token):
if user.email_verified:
flash('すでに認証済みです。', 'info')
else:
user.email_verified = True
db.session.commit()
flash('メール認証が完了しました!', 'success')
return redirect(url_for('login'))
ログイン画面でログイン時、メール認証できてなかったらメッセージ表示
if not user.email_verified:
flash('メール認証が完了していません。', 'error')
return redirect(url_for('login'))
パスワリセットボタンを設定
class PasswordResetRequestForm(FlaskForm):
submit = SubmitField('パスワードリセットリンク送信')
パスワリセットボタンが押されたら、tokenとリセットリンクを送り、message"リセットリンクを送りました"と表示、ログインページへ行かせる
@app.route('/reset_password', methods=['GET', 'POST'])
def reset_request():
form = PasswordResetRequestForm()
if form.validate_on_submit():
user = User.query.filter_by(email=form.email.data).first()
if user:
token = s.dumps(user.email, salt='password-reset')
link = url_for('reset_token', token=token, _external=True)
msg = Message('パスワードリセット', sender='あなたのメールアドレス', recipients=[user.email])
msg.body = f'パスワードリセットはこちら: {link}'
mail.send(msg)
flash('リセットリンクを送信しました。メールを確認してください。', 'success')
return redirect(url_for('login'))
[パスワリクエストページ]パスワリクエストページ処理と処理ページ作成
app.py
@app.route('/reset_password', methods=['GET', 'POST'])
def reset_request():
form = PasswordResetRequestForm()
if form.validate_on_submit():
user = User.query.filter_by(email=form.email.data).first()
if user:
token = s.dumps(user.email, salt='password-reset')
link = url_for('reset_token', token=token, _external=True)
msg = Message('パスワードリセット', sender='あなたのメールアドレス', recipients=[user.email])
msg.body = f'パスワードリセットはこちら: {link}'
mail.send(msg)
flash('リセットリンクを送信しました。メールを確認してください。', 'success')
return redirect(url_for('login'))
return render_template('reset_request.html', form=form)
reset_request.html
<!DOCTYPE html>
<html lang="ja">
<head>
<meta charset="UTF-8">
<title>パスワードリセットリクエスト</title>
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css" rel="stylesheet">
</head>
<body class="bg-light">
<div class="container mt-5">
<div class="row justify-content-center">
<div class="col-md-6">
<h2 class="text-center">パスワードリセット</h2>
<form method="POST">
{{ form.hidden_tag() }}
<div class="mb-3">
{{ form.email.label(class="form-label") }}
{{ form.email(class="form-control") }}
{% for error in form.email.errors %}
<div class="text-danger">{{ error }}</div>
{% endfor %}
</div>
<div class="d-grid">
{{ form.submit(class="btn btn-primary") }}
</div>
</form>
<div class="mt-3 text-center">
<a href="{{ url_for('login') }}">ログイン画面に戻る</a>
</div>
</div>
</div>
</div>
</body>
</html>
パスワードリセット画面でtokenを1時間保存(保持)、token切れならリセット画面へ行かせ、tokenとリンク再発行
@app.route('/reset_password/<token>', methods=['GET', 'POST'])
def reset_token(token):
try:
email = s.loads(token, salt='password-reset', max_age=3600)
except SignatureExpired:
flash('リンクの有効期限が切れています。', 'error')
return redirect(url_for('reset_request'))
新パスワ入力ページ作成
reset_token.html
<!DOCTYPE html>
<html lang="ja">
<head>
<meta charset="UTF-8">
<title>パスワードリセット</title>
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css" rel="stylesheet">
</head>
<body class="bg-light">
<div class="container mt-5">
<div class="row justify-content-center">
<div class="col-md-6">
<h2 class="text-center">新しいパスワード設定</h2>
<form method="POST">
{{ form.hidden_tag() }}
<div class="mb-3">
{{ form.password.label(class="form-label") }}
{{ form.password(class="form-control") }}
{% for error in form.password.errors %}
<div class="text-danger">{{ error }}</div>
{% endfor %}
</div>
<div class="d-grid">
{{ form.submit(class="btn btn-success", value="パスワードリセット") }}
</div>
</form>
<div class="mt-3 text-center">
<a href="{{ url_for('login') }}">ログイン画面に戻る</a>
</div>
</div>
</div>
</div>
</body>
</html>
tokenを保持/ token再発行する必要性なくなったら、パスワードをもう一度登録し、ログイン画面へ
@app.route('/reset_password/<token>', methods=['GET', 'POST'])
def reset_token(token):
user = User.query.filter_by(email=email).first_or_404()
form = RegisterForm()
if form.validate_on_submit():
hashed_pw = bcrypt.generate_password_hash(form.password.data).decode('utf-8')
user.password = hashed_pw
db.session.commit()
flash('パスワードをリセットしました。', 'success')
return redirect(url_for('login'))
メール認証準備
AzureポータルでOutlook用"client_secret.json"作る
1️⃣ 左メニューAzure Active Directory → アプリの登録 → 新規登録
→アプリ名入力 → リダイレクトURIは空白 → 登録 → クライアント)IDコピペ
2️⃣ 左メニュー"証明書とシークレット"
→ 新しいクライアントシークレット
→ 説明:「Flask用」
→ 有効期限:24か月
→ 値」をすぐコピペ
3️⃣ 左メニュー「APIのアクセス許可」
→ アクセス許可の追加
→ Microsoft Graph
→ 委任されたアクセス許可
→ SMTP.Send にチェック
→ アクセス許可を付与」ボタン
4️⃣ client_secret.json作成
azure リンク ↓
4️⃣ client_secret.json作成、Flaskアプリと同じフォルダに置く
{
"installed": {
"client_id": "さっき控えたアプリID",
"client_secret": "さっき控えたシークレット",
"auth_uri": "https://login.microsoftonline.com/common/oauth2/v2.0/authorize",
"token_uri": "https://login.microsoftonline.com/common/oauth2/v2.0/token",
"redirect_uris": [
"http://localhost"
]
}
}
outlook Oathメール認証 下準備
pip install Flask requests requests_oauthlib
outlook Oathメール認証 code
app,py
from flask import Flask, redirect, url_for, session, request
from requests_oauthlib import OAuth2Session
import os
import requests
app = Flask(__name__)
app.secret_key = os.environ.get("FLASK_SECRET_KEY", os.urandom(24)) # 本番は環境変数から読む!
# ===== OAuth情報(全部環境変数に入れる!) =====
CLIENT_ID = os.environ.get("OAUTH_CLIENT_ID")
CLIENT_SECRET = os.environ.get("OAUTH_CLIENT_SECRET")
AUTHORIZATION_BASE_URL = "https://login.microsoftonline.com/common/oauth2/v2.0/authorize"
TOKEN_URL = "https://login.microsoftonline.com/common/oauth2/v2.0/token"
REDIRECT_URI = os.environ.get("OAUTH_REDIRECT_URI", "http://localhost:5000/callback")
SCOPE = ["https://graph.microsoft.com/Mail.Send"]
@app.route('/')
def login():
outlook = OAuth2Session(CLIENT_ID, redirect_uri=REDIRECT_URI, scope=SCOPE)
authorization_url, state = outlook.authorization_url(AUTHORIZATION_BASE_URL)
session['oauth_state'] = state
return redirect(authorization_url)
@app.route('/callback')
def callback():
outlook = OAuth2Session(CLIENT_ID, state=session['oauth_state'], redirect_uri=REDIRECT_URI)
try:
token = outlook.fetch_token(TOKEN_URL, client_secret=CLIENT_SECRET, authorization_response=request.url)
send_email(token)
return '認証成功!メールを送信しました!'
except Exception as e:
print('エラー:', e)
return '認証または送信失敗…'
def send_email(token):
access_token = token['access_token']
headers = {
'Authorization': f'Bearer {access_token}',
'Content-Type': 'application/json'
}
email_msg = {
"message": {
"subject": "【認証メール】あなたの登録を確認しました!",
"body": {
"contentType": "Text",
"content": "Outlook認証が成功しました。ログインが完了しました!"
},
"toRecipients": [
{
"emailAddress": {
"address": os.environ.get("YOUR_OUTLOOK_EMAIL")
}
}
]
}
}
response = requests.post(
'https://graph.microsoft.com/v1.0/me/sendMail',
headers=headers,
json=email_msg
)
if response.status_code != 202:
raise Exception('メール送信失敗: ' + response.text)
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000, debug=False)
envファイルにまとめとくと吉
FLASK_SECRET_KEY Flask用のランダム文字列
OAUTH_CLIENT_ID Azure client_id
OAUTH_CLIENT_SECRET Azure client_secret
OAUTH_REDIRECT_URI 本番サーバーURL(例:https://yourdomain.com/callback)
YOUR_OUTLOOK_EMAIL Outlookアドレス
メール認証できたら、トークン保存
import json
@app.route('/callback')
def callback():
outlook = OAuth2Session(CLIENT_ID, state=session['oauth_state'], redirect_uri=REDIRECT_URI)
try:
token = outlook.fetch_token(TOKEN_URL, client_secret=CLIENT_SECRET, authorization_response=request.url)
# 👇ここを追加!
with open('token.json', 'w') as f:
json.dump(token, f)
send_email(token)
return '認証成功!メールを送信しました!'
except Exception as e:
print('エラー:', e)
return '認証または送信失敗…'
メール認証一回やったら、token.jsonからトークン読み込み、メール再認証しない[これがないと、毎回メール再認証し続けなきゃいけなくなる]
import json
def read_token_from_file():
try:
with open('token.json', 'r') as f:
token = json.load(f)
return token
except FileNotFoundError:
print("token.json が見つかりません")
return None
:::token.jsonがない場合、新トークン取得
def get_token():
# トークンがファイルにあれば、それを使う
token = read_token_from_file()
if not token:
# トークンがない場合、OAuth認証を行って新しいトークンを取得
outlook = OAuth2Session(CLIENT_ID, redirect_uri=REDIRECT_URI, scope=SCOPE)
authorization_url, state = outlook.authorization_url(AUTHORIZATION_BASE_URL)
session['oauth_state'] = state
return redirect(authorization_url)
return token
callback関数 修正
@app.route('/callback')
def callback():
token = get_token()
if token is None:
return '認証エラー'
send_email(token)
return '認証成功!メールを送信しました!'
メール認証できたら「ユーザー認証OKフラグ」をSqlite DBに記録
import sqlite3
conn = sqlite3.connect('users.db')
c = conn.cursor()
c.execute('CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, email TEXT, verified INTEGER)')
conn.commit()
conn.close()
フラグを保存したら、SqliteDB更新
def save_verified_user(email):
conn = sqlite3.connect('users.db')
c = conn.cursor()
c.execute('INSERT INTO users (email, verified) VALUES (?, ?)', (email, 1))
conn.commit()
conn.close()
def send_email(token):
...
if response.status_code == 202:
save_verified_user(os.environ.get("YOUR_OUTLOOK_EMAIL"))
else:
raise Exception('メール送信失敗: ' + response.text)
リンクをSSL化する
@app.before_request
def before_request():
if not request.is_secure and not app.debug:
url = request.url.replace("http://", "https://", 1)
return redirect(url)
ただのルール
メール欄に何か入力されてたら : form.email.data
ボタンが押され、バリデーションがok(エラーなし)なら : form.validate_on_submit()
メアドで認証 / ログイン / 新規登録してるか調べる
user = User.query.filter_by(email=form.email.data).first()
ユーザー名で認証 / ログイン / 新規登録してるか調べる
user = User.query.filter_by(username=form.username.data).first()
if文 全部にreturnついてないと、redirect(url?for())と書いててもリダイレクトされない
例文(before)
form = RegisterForm()
if form.validate_on_submit():
hashed_pw = bcrypt.generate_password_hash(form.password.data).decode('utf-8')
# メールアドレスが入力されていたら「メール認証あり」
if form.email.data:
user = User(username=form.username.data, email=form.email.data, password=hashed_pw)
db.session.add(user)
db.session.commit()
token = os.dumps(user.email, salt='email-confirm')
link = url_for('confirm_email', token=token, _external=True)
msg = Message('メール認証を完了してください', sender='あなたのメールアドレス', recipients=[user.email])
msg.body = f'こちらのリンクをクリックして認証を完了してください: {link}'
mail.send(msg)
flash('認証メールを送りました。メールを確認してください!', 'success')
# ここだけ、returnない!!
# メールアドレスなし → 通常登録
else:
new_user = User(username=form.username.data, password=hashed_pw)
db.session.add(new_user)
db.session.commit()
flash('登録が完了しました!ログインしてください。', 'success')
return redirect(url_for('register_complete'))
return render_template('register.html', form=form)
after
if form.validate_on_submit():
hashed_pw = bcrypt.generate_password_hash(form.password.data).decode('utf-8')
if form.email.data:
user = User(username=form.username.data, email=form.email.data, password=hashed_pw)
db.session.add(user)
db.session.commit()
s = URLSafeTimedSerializer(app.config['SECRET_KEY'])
token = os.dumps(user.email, salt='email-confirm')
link = url_for('confirm_email', token=token, _external=True)
msg = Message('メール認証を完了してください', sender='あなたのメールアドレス', recipients=[user.email])
msg.body = f'こちらのリンクをクリックして認証を完了してください: {link}'
mail.send(msg)
flash('認証メールを送りました。メールを確認してください!', 'success')
else:
new_user = User(username=form.username.data, password=hashed_pw)
db.session.add(new_user)
db.session.commit()
flash('登録が完了しました!ログインしてください。', 'success')
return redirect(url_for('register_complete')) # ← メールあり・なし共通で遷移
return render_template('register.html', form=form)
[バリデーション]メールアドレスが無しでも登録したい
from wtforms.validators import Optional
class RegisterForm(FlaskForm):
email = StringField('Email', validators=[Optional(), Email()])
def validate_email(self, email):
if not email.data:
return # メールが空なら重複チェックしない
user = User.query.filter_by(email=email.data).first()
if user:
raise ValidationError('このメールアドレスは既に登録されています')
✍️Userモデルを変更
class User(db.Model, UserMixin):
email = db.Column(db.String(120), unique=True, nullable=True)
コマンドでDB更新 準備
pip install flask-migrate
コマンドでDB更新 設定追加
1.app.pyにこれ書く
from flask_migrate import Migrate
db = SQLAlchemy(app)
migrate = Migrate(app, db)
2 . DBファイル消して、初期化コマンド実行
flask run --reload
rm -rf migrations/
rm site.db
flask db init
flask db migrate -m "Allow email to be nullable"
flask db upgrade
# 結果確認
flask db current
Flaskアプリ再起動!
flask db currentがこうなってたらDBできてる
INFO [alembic.runtime.migration] Context impl SQLiteImpl.
INFO [alembic.runtime.migration] Will assume non-transactional DDL.
21ecf29caf10 (head)