/
Evdik
/
Integration_System_labs
Обзор
Документация
Войти
/
Evdik
/
Integration_System_labs
Код
Запросы
1
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
app.py
999 строк
32 KB
Evdikkom
2_laba
15 окт 2025, 01:43
15 окт 2025, 01:43
1c003f8
Код
Авторство
О чём код?
from flask import Flask, request, jsonify from flask_sqlalchemy import SQLAlchemy from flask_jwt_extended import JWTManager, create_access_token, jwt_required, get_jwt_identity from flask_limiter import Limiter from flask_limiter.util import get_remote_address from flasgger import Swagger from datetime import datetime, timedelta import uuid from decimal import Decimal from functools import wraps import os import hashlib app = Flask(__name__) app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///gamestore.db' app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False app.config['JWT_SECRET_KEY'] = 'your-super-secret-game-store-key' app.config['JWT_ACCESS_TOKEN_EXPIRES'] = timedelta(hours=1) # Внутренний API ключ (в продакшене хранить в переменных окружения) app.config['INTERNAL_API_KEY'] = 'internal-secret-key-12345' db = SQLAlchemy(app) jwt = JWTManager(app) # Улучшенный rate limiting с заголовками limiter = Limiter( app=app, key_func=get_remote_address, default_limits=["200 per day", "50 per hour"], headers_enabled=True ) # Обновленная конфигурация Swagger app.config['SWAGGER'] = { 'title': 'Game Store API v2', 'uiversion': 3, 'specs_route': '/docs/', 'specs': [ { 'endpoint': 'apispec_1', 'route': '/apispec_1.json', } ] } swagger = Swagger(app) ###################### Модели данных (обновленные) ####################### class User(db.Model): id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(80), unique=True, nullable=False) email = db.Column(db.String(120), unique=True, nullable=False) password_hash = db.Column(db.String(128), nullable=False) balance = db.Column(db.Float, default=0.0) role = db.Column(db.String(20), default='user') is_active = db.Column(db.Boolean, default=True) last_login = db.Column(db.DateTime) orders = db.relationship('Order', backref='user', lazy=True) reviews = db.relationship('Review', backref='user', lazy=True) created_at = db.Column(db.DateTime, default=datetime.utcnow) updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) class Game(db.Model): id = db.Column(db.Integer, primary_key=True) title = db.Column(db.String(100), nullable=False) description = db.Column(db.Text) short_description = db.Column(db.String(200)) price = db.Column(db.Float, nullable=False) original_price = db.Column(db.Float) developer = db.Column(db.String(100)) publisher = db.Column(db.String(100)) release_date = db.Column(db.Date) genre = db.Column(db.String(50)) platform = db.Column(db.String(50)) is_available = db.Column(db.Boolean, default=True) is_featured = db.Column(db.Boolean, default=False) stock = db.Column(db.Integer, default=-1) image_url = db.Column(db.String(200)) trailer_url = db.Column(db.String(200)) system_requirements = db.Column(db.Text) created_at = db.Column(db.DateTime, default=datetime.utcnow) updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) order_items = db.relationship('OrderItem', backref='game', lazy=True) game_keys = db.relationship('GameKey', backref='game', lazy=True) reviews = db.relationship('Review', backref='game', lazy=True) class Order(db.Model): id = db.Column(db.Integer, primary_key=True) user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False) total_amount = db.Column(db.Float, nullable=False) status = db.Column(db.String(20), default='pending') payment_method = db.Column(db.String(50)) created_at = db.Column(db.DateTime, default=datetime.utcnow) updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) order_items = db.relationship('OrderItem', backref='order', lazy=True) class OrderItem(db.Model): id = db.Column(db.Integer, primary_key=True) order_id = db.Column(db.Integer, db.ForeignKey('order.id'), nullable=False) game_id = db.Column(db.Integer, db.ForeignKey('game.id'), nullable=False) quantity = db.Column(db.Integer, default=1) price_at_purchase = db.Column(db.Float, nullable=False) game_key = db.relationship('GameKey', backref='order_item', uselist=False) class GameKey(db.Model): id = db.Column(db.Integer, primary_key=True) game_id = db.Column(db.Integer, db.ForeignKey('game.id'), nullable=False) key = db.Column(db.String(100), unique=True, nullable=False) is_used = db.Column(db.Boolean, default=False) order_item_id = db.Column(db.Integer, db.ForeignKey('order_item.id'), nullable=True) created_at = db.Column(db.DateTime, default=datetime.utcnow) class Review(db.Model): id = db.Column(db.Integer, primary_key=True) user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False) game_id = db.Column(db.Integer, db.ForeignKey('game.id'), nullable=False) rating = db.Column(db.Integer, nullable=False) comment = db.Column(db.Text) is_verified = db.Column(db.Boolean, default=False) created_at = db.Column(db.DateTime, default=datetime.utcnow) updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) class IdempotencyKey(db.Model): id = db.Column(db.Integer, primary_key=True) key = db.Column(db.String(36), unique=True, nullable=False) user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False) resource_path = db.Column(db.String(200), nullable=False) response_code = db.Column(db.Integer, nullable=False) response_body = db.Column(db.Text, nullable=False) created_at = db.Column(db.DateTime, default=datetime.utcnow) # Создание таблиц with app.app_context(): db.create_all() ########################## Вспомогательные функции ############################### def hash_password(password): """Безопасное хеширование пароля""" return hashlib.sha256(password.encode()).hexdigest() def verify_password(password, password_hash): """Проверка пароля""" return hash_password(password) == password_hash def check_internal_api_key(): """Проверка внутреннего API ключа""" api_key = request.headers.get('X-API-Key') return api_key == app.config['INTERNAL_API_KEY'] def internal_api_required(f): """Декоратор для внутренних API endpoints""" @wraps(f) def decorated_function(*args, **kwargs): if not check_internal_api_key(): return jsonify({'error': 'Invalid internal API key'}), 401 return f(*args, **kwargs) return decorated_function def apply_pagination(query, page, per_page): """Применение пагинации к запросу""" page = max(1, page) per_page = min(100, max(1, per_page)) # Ограничение до 100 на страницу paginated = query.paginate( page=page, per_page=per_page, error_out=False ) return { 'items': paginated.items, 'total': paginated.total, 'pages': paginated.pages, 'current_page': paginated.page, 'per_page': paginated.per_page, 'has_next': paginated.has_next, 'has_prev': paginated.has_prev } def filter_fields(data, fields): """Фильтрация полей на основе параметра fields""" if not fields: return data if isinstance(data, list): return [{k: v for k, v in item.items() if k in fields} for item in data] else: return {k: v for k, v in data.items() if k in fields} ########################## Аутентификация ############################### @app.route('/api/v1/auth/register', methods=['POST']) @limiter.limit("5 per minute") def register(): """ Регистрация пользователя --- tags: - Auth parameters: - in: body name: body required: true schema: type: object required: - username - email - password properties: username: type: string email: type: string password: type: string responses: 201: description: Пользователь создан 400: description: Ошибка валидации """ data = request.get_json() if User.query.filter_by(username=data['username']).first(): return jsonify({'error': 'Username already exists'}), 400 if User.query.filter_by(email=data['email']).first(): return jsonify({'error': 'Email already exists'}), 400 user = User( username=data['username'], email=data['email'], password_hash=hash_password(data['password']) ) db.session.add(user) db.session.commit() return jsonify({'message': 'User created successfully'}), 201 @app.route('/api/v1/auth/login', methods=['POST']) @limiter.limit("10 per minute") def login(): """ Аутентификация пользователя --- tags: - Auth parameters: - in: body name: body required: true schema: type: object required: - username - password properties: username: type: string password: type: string responses: 200: description: Успешный вход 401: description: Неверные учетные данные """ data = request.get_json() user = User.query.filter_by(username=data['username']).first() if user and verify_password(data['password'], user.password_hash): user.last_login = datetime.utcnow() db.session.commit() access_token = create_access_token(identity={ 'id': user.id, 'username': user.username, 'role': user.role }) return jsonify({ 'access_token': access_token, 'user': { 'id': user.id, 'username': user.username, 'email': user.email, 'balance': user.balance, 'role': user.role } }), 200 return jsonify({'error': 'Invalid credentials'}), 401 ########################## Пользователь ############################### @app.route('/api/v1/user/profile', methods=['GET']) @jwt_required() def get_profile(): """ Получить профиль пользователя --- tags: - User security: - Bearer: [] parameters: - name: fields in: query type: string required: false description: Поля для включения в ответ (через запятую) responses: 200: description: Профиль пользователя """ current_user = get_jwt_identity() user = User.query.get(current_user['id']) fields_param = request.args.get('fields') fields = fields_param.split(',') if fields_param else None profile_data = { 'id': user.id, 'username': user.username, 'email': user.email, 'balance': user.balance, 'role': user.role, 'is_active': user.is_active, 'last_login': user.last_login.isoformat() if user.last_login else None, 'created_at': user.created_at.isoformat(), 'updated_at': user.updated_at.isoformat() } filtered_data = filter_fields(profile_data, fields) return jsonify(filtered_data), 200 ########################## Игры с пагинацией и опциональными полями ############################### @app.route('/api/v1/games', methods=['GET']) def get_games(): """ Получить каталог игр с пагинацией --- tags: - Games parameters: - name: page in: query type: integer required: false default: 1 - name: per_page in: query type: integer required: false default: 10 - name: genre in: query type: string required: false - name: platform in: query type: string required: false - name: available in: query type: boolean required: false - name: fields in: query type: string required: false description: Поля для включения в ответ (через запятую) responses: 200: description: Список игр с пагинацией """ page = request.args.get('page', 1, type=int) per_page = request.args.get('per_page', 10, type=int) genre = request.args.get('genre') platform = request.args.get('platform') available = request.args.get('available', type=bool) fields_param = request.args.get('fields') fields = fields_param.split(',') if fields_param else None query = Game.query if genre: query = query.filter(Game.genre == genre) if platform: query = query.filter(Game.platform == platform) if available is not None: query = query.filter(Game.is_available == available) # Применяем пагинацию pagination_data = apply_pagination(query, page, per_page) games_data = [] for game in pagination_data['items']: game_info = { 'id': game.id, 'title': game.title, 'short_description': game.short_description, 'price': game.price, 'original_price': game.original_price, 'developer': game.developer, 'genre': game.genre, 'platform': game.platform, 'is_available': game.is_available, 'is_featured': game.is_featured, 'image_url': game.image_url, 'release_date': game.release_date.isoformat() if game.release_date else None, 'created_at': game.created_at.isoformat() } games_data.append(game_info) filtered_games = filter_fields(games_data, fields) response = { 'games': filtered_games, 'pagination': { 'total': pagination_data['total'], 'pages': pagination_data['pages'], 'current_page': pagination_data['current_page'], 'per_page': pagination_data['per_page'], 'has_next': pagination_data['has_next'], 'has_prev': pagination_data['has_prev'] } } return jsonify(response), 200 @app.route('/api/v1/games/<int:game_id>', methods=['GET']) def get_game(game_id): """ Получить детали игры с опциональными полями --- tags: - Games parameters: - name: game_id in: path type: integer required: true - name: fields in: query type: string required: false description: Поля для включения в ответ (через запятую) responses: 200: description: Данные игры 404: description: Игра не найдена """ game = Game.query.get_or_404(game_id) fields_param = request.args.get('fields') fields = fields_param.split(',') if fields_param else None game_data = { 'id': game.id, 'title': game.title, 'description': game.description, 'short_description': game.short_description, 'price': game.price, 'original_price': game.original_price, 'developer': game.developer, 'publisher': game.publisher, 'genre': game.genre, 'platform': game.platform, 'release_date': game.release_date.isoformat() if game.release_date else None, 'is_available': game.is_available, 'is_featured': game.is_featured, 'stock': game.stock, 'image_url': game.image_url, 'trailer_url': game.trailer_url, 'system_requirements': game.system_requirements, 'created_at': game.created_at.isoformat(), 'updated_at': game.updated_at.isoformat(), 'reviews_count': len(game.reviews), 'average_rating': sum(r.rating for r in game.reviews) / len(game.reviews) if game.reviews else 0 } filtered_data = filter_fields(game_data, fields) return jsonify(filtered_data), 200 ########################## Заказы с пагинацией ############################### def idempotent_request(f): """Декоратор идемпотентности""" @wraps(f) def decorated_function(*args, **kwargs): if request.method == 'POST': idempotency_key = request.headers.get('Idempotency-Key') current_user = get_jwt_identity() resource_path = request.path if idempotency_key: existing_key = IdempotencyKey.query.filter_by( key=idempotency_key, user_id=current_user['id'], resource_path=resource_path ).first() if existing_key: return jsonify(existing_key.response_body), existing_key.response_code response = f(*args, **kwargs) if request.method == 'POST' and idempotency_key and 200 <= response[1] < 300: new_key = IdempotencyKey( key=idempotency_key, user_id=current_user['id'], resource_path=resource_path, response_code=response[1], response_body=response[0].get_data(as_text=True) ) db.session.add(new_key) db.session.commit() return response return decorated_function @app.route('/api/v1/orders', methods=['POST']) @jwt_required() @idempotent_request @limiter.limit("10 per minute") def create_order(): """ Создать заказ --- tags: - Orders security: - Bearer: [] parameters: - in: body name: body required: true schema: type: object required: - items properties: items: type: array items: type: object properties: game_id: type: integer quantity: type: integer responses: 201: description: Заказ создан 400: description: Ошибка валидации """ current_user = get_jwt_identity() data = request.get_json() user = User.query.get(current_user['id']) total_amount = 0 order_items = [] for item in data['items']: game = Game.query.get(item['game_id']) if not game or not game.is_available: return jsonify({'error': f'Game {item["game_id"]} not available'}), 400 if game.stock != -1 and game.stock < item['quantity']: return jsonify({'error': f'Not enough stock for {game.title}'}), 400 total_amount += game.price * item['quantity'] if user.balance < total_amount: return jsonify({'error': 'Insufficient balance'}), 400 order = Order( user_id=user.id, total_amount=total_amount, status='pending' ) db.session.add(order) for item in data['items']: game = Game.query.get(item['game_id']) order_item = OrderItem( order=order, game=game, quantity=item['quantity'], price_at_purchase=game.price ) db.session.add(order_item) if game.stock != -1: available_key = GameKey.query.filter_by( game_id=game.id, is_used=False ).first() if available_key: available_key.order_item = order_item available_key.is_used = True if game.stock > 0: game.stock -= 1 user.balance -= total_amount order.status = 'completed' db.session.commit() return jsonify({ 'order_id': order.id, 'total_amount': total_amount, 'status': order.status, 'message': 'Order completed successfully' }), 201 @app.route('/api/v1/orders', methods=['GET']) @jwt_required() def get_orders(): """ Получить историю заказов с пагинацией --- tags: - Orders security: - Bearer: [] parameters: - name: page in: query type: integer required: false default: 1 - name: per_page in: query type: integer required: false default: 10 responses: 200: description: Список заказов с пагинацией """ current_user = get_jwt_identity() page = request.args.get('page', 1, type=int) per_page = request.args.get('per_page', 10, type=int) query = Order.query.filter_by(user_id=current_user['id']) pagination_data = apply_pagination(query, page, per_page) orders_data = [] for order in pagination_data['items']: order_info = { 'id': order.id, 'total_amount': order.total_amount, 'status': order.status, 'payment_method': order.payment_method, 'created_at': order.created_at.isoformat(), 'updated_at': order.updated_at.isoformat(), 'items_count': len(order.order_items), 'items': [{ 'game_id': item.game_id, 'game_title': item.game.title, 'quantity': item.quantity, 'price': item.price_at_purchase } for item in order.order_items] } orders_data.append(order_info) response = { 'orders': orders_data, 'pagination': { 'total': pagination_data['total'], 'pages': pagination_data['pages'], 'current_page': pagination_data['current_page'], 'per_page': pagination_data['per_page'], 'has_next': pagination_data['has_next'], 'has_prev': pagination_data['has_prev'] } } return jsonify(response), 200 ########################## Баланс и ключи ############################### @app.route('/api/v1/user/deposit', methods=['POST']) @jwt_required() @idempotent_request def deposit_balance(): """ Пополнить баланс --- tags: - User security: - Bearer: [] parameters: - in: body name: body required: true schema: type: object required: - amount properties: amount: type: number responses: 200: description: Баланс пополнен """ current_user = get_jwt_identity() data = request.get_json() user = User.query.get(current_user['id']) user.balance += float(data['amount']) db.session.commit() return jsonify({ 'new_balance': user.balance, 'message': 'Balance updated successfully' }), 200 ########################## Отзывы с пагинацией ############################### @app.route('/api/v2/games/<int:game_id>/reviews', methods=['POST']) @jwt_required() def create_review(game_id): """ Добавить отзыв к игре --- tags: - Reviews security: - Bearer: [] parameters: - name: game_id in: path type: integer required: true - in: body name: body required: true schema: type: object required: - rating properties: rating: type: integer comment: type: string responses: 201: description: Отзыв добавлен 400: description: Ошибка валидации """ current_user = get_jwt_identity() data = request.get_json() has_purchased = OrderItem.query.join(Order).filter( Order.user_id == current_user['id'], OrderItem.game_id == game_id, Order.status == 'completed' ).first() if not has_purchased: return jsonify({'error': 'You must purchase the game before reviewing'}), 400 existing_review = Review.query.filter_by( user_id=current_user['id'], game_id=game_id ).first() if existing_review: return jsonify({'error': 'You have already reviewed this game'}), 400 review = Review( user_id=current_user['id'], game_id=game_id, rating=data['rating'], comment=data.get('comment', ''), is_verified=True # Помечаем как проверенный, так как пользователь купил игру ) db.session.add(review) db.session.commit() return jsonify({ 'id': review.id, 'message': 'Review added successfully' }), 201 @app.route('/api/v2/games/<int:game_id>/reviews', methods=['GET']) def get_game_reviews(game_id): """ Получить отзывы об игре с пагинацией --- tags: - Reviews parameters: - name: game_id in: path type: integer required: true - name: page in: query type: integer required: false default: 1 - name: per_page in: query type: integer required: false default: 10 responses: 200: description: Список отзывов с пагинацией """ page = request.args.get('page', 1, type=int) per_page = request.args.get('per_page', 10, type=int) query = Review.query.filter_by(game_id=game_id) pagination_data = apply_pagination(query, page, per_page) reviews_data = [] for review in pagination_data['items']: review_info = { 'id': review.id, 'user_id': review.user_id, 'username': review.user.username, 'rating': review.rating, 'comment': review.comment, 'is_verified': review.is_verified, 'created_at': review.created_at.isoformat(), 'updated_at': review.updated_at.isoformat() } reviews_data.append(review_info) response = { 'reviews': reviews_data, 'pagination': { 'total': pagination_data['total'], 'pages': pagination_data['pages'], 'current_page': pagination_data['current_page'], 'per_page': pagination_data['per_page'], 'has_next': pagination_data['has_next'], 'has_prev': pagination_data['has_prev'] } } return jsonify(response), 200 ########################## ВНУТРЕННИЕ API ############################### @app.route('/internal/stats', methods=['GET']) @internal_api_required def get_internal_stats(): """ Внутренняя статистика системы --- tags: - Internal parameters: - name: period in: query type: string required: false description: Период для статистики (day, week, month) responses: 200: description: Статистика системы 401: description: Неверный внутренний API ключ """ period = request.args.get('period', 'day') # Упрощенная статистика (в реальном проекте были бы сложные запросы) total_users = User.query.count() total_games = Game.query.count() total_orders = Order.query.count() total_revenue = db.session.query(db.func.sum(Order.total_amount)).scalar() or 0 # Активные пользователи за последний месяц (упрощенно) month_ago = datetime.utcnow() - timedelta(days=30) active_users = User.query.filter(User.last_login >= month_ago).count() # Популярные жанры (упрощенно) genre_stats = db.session.query( Game.genre, db.func.count(Game.id).label('count') ).group_by(Game.genre).all() stats = { 'period': period, 'summary': { 'total_users': total_users, 'active_users': active_users, 'total_games': total_games, 'total_orders': total_orders, 'total_revenue': float(total_revenue) }, 'genres': [{'genre': genre, 'count': count} for genre, count in genre_stats], 'timestamp': datetime.utcnow().isoformat() } return jsonify(stats), 200 @app.route('/internal/users/<int:user_id>/debug', methods=['GET']) @internal_api_required def get_user_debug_info(user_id): """ Отладочная информация о пользователе (внутренний API) --- tags: - Internal parameters: - name: user_id in: path type: integer required: true responses: 200: description: Отладочная информация 404: description: Пользователь не найден """ user = User.query.get_or_404(user_id) # Расширенная информация для внутреннего использования user_orders = Order.query.filter_by(user_id=user_id).all() user_reviews = Review.query.filter_by(user_id=user_id).all() debug_info = { 'user': { 'id': user.id, 'username': user.username, 'email': user.email, 'balance': user.balance, 'role': user.role, 'is_active': user.is_active, 'created_at': user.created_at.isoformat(), 'last_login': user.last_login.isoformat() if user.last_login else None }, 'orders': { 'total': len(user_orders), 'total_spent': sum(order.total_amount for order in user_orders), 'recent_orders': [{ 'id': order.id, 'total_amount': order.total_amount, 'status': order.status, 'created_at': order.created_at.isoformat() } for order in user_orders[:5]] # Только последние 5 заказов }, 'reviews': { 'total': len(user_reviews), 'average_rating': sum(r.rating for r in user_reviews) / len(user_reviews) if user_reviews else 0, 'recent_reviews': [{ 'game_id': review.game_id, 'game_title': review.game.title, 'rating': review.rating, 'created_at': review.created_at.isoformat() } for review in user_reviews[:5]] # Только последние 5 отзывов } } return jsonify(debug_info), 200 ########################## Обработчики ошибок ############################### @app.errorhandler(429) def ratelimit_handler(e): """Обработчик превышения лимита запросов""" return jsonify({ 'error': 'Rate limit exceeded', 'message': str(e.description) }), 429, { 'Retry-After': e.description.split(' ')[3], 'X-RateLimit-Remaining': '0' } # Middleware для добавления заголовков rate limiting @app.after_request def add_rate_limit_headers(response): """Добавляет заголовки rate limiting к каждому ответу""" # Flask-Limiter автоматически добавляет эти заголовки, # но мы можем их кастомизировать если нужно return response if __name__ == '__main__': app.run(debug=True)