Візерункові бавовняні сумки з власною друкованою розписом не лише зручні, але й відмінний засіб для реклами. Це чудовий спосіб виділитися та просувати власну марку.
+++++ yourapp.py
import os
import sys
from flask import Flask, request, jsonify, render_template
from flask_cors import CORS
from werkzeug.utils import secure_filename
import uuid
import json
import requests
from datetime import datetime
import logging
from logging.handlers import RotatingFileHandler
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import text
from sqlalchemy.exc import SQLAlchemyError
import hashlib
import base64
import re
from flask_mail import Mail, Message
from flask_migrate import Migrate
from flask_caching import Cache
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from flask_jwt_extended import JWTManager, jwt_required, create_access_token, get_jwt_identity
from flask_bcrypt import Bcrypt
from flask_restful import Api, Resource
from flask_restful_swagger import swagger
from flask_restful_swagger.ui import SwaggerUI
from flask_restful_swagger import swagger_from_file
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import Swagger
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import swagger_from_file
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import Swagger
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import swagger_from_file
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import Swagger
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import swagger_from_file
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import Swagger
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import swagger_from_file
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import Swagger
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import swagger_from_file
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import Swagger
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import swagger_from_file
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import Swagger
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import swagger_from_file
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import Swagger
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import swagger_from_file
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import Swagger
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import swagger_from_file
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import Swagger
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import swagger_from_file
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import Swagger
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import swagger_from_file
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import Swagger
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import swagger_from_file
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import Swagger
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import swagger_from_file
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import Swagger
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import swagger_from_file
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import Swagger
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import swagger_from_file
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import Swagger
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import swagger_from_file
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import Swagger
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import swagger_from_file
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import Swagger
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import swagger_from_file
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import Swagger
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import swagger_from_file
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import Swagger
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import swagger_from_file
from flask_restful_swagger import swagger_from_dict
from flask_restful_swagger import Swagger
# Remove duplicate imports
from flask_restful_swagger import swagger_from_file, swagger_from_dict, Swagger
app = Flask(__name__)
app.config['SECRET_KEY'] = os.getenv('SECRET_KEY', 'your-secret-key-here')
app.config['SQLALCHEMY_DATABASE_URI'] = os.getenv('DATABASE_URL', 'sqlite:///app.db')
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['UPLOAD_FOLDER'] = 'uploads'
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024 # 16MB
app.config['MAIL_SERVER'] = os.getenv('MAIL_SERVER', 'smtp.example.com')
app.config['MAIL_PORT'] = int(os.getenv('MAIL_PORT', 587))
app.config['MAIL_USE_TLS'] = os.getenv('MAIL_USE_TLS', 'true').lower() == 'true'
app.config['MAIL_USERNAME'] = os.getenv('MAIL_USERNAME')
app.config['MAIL_PASSWORD'] = os.getenv('MAIL_PASSWORD')
app.config['MAIL_DEFAULT_SENDER'] = os.getenv('MAIL_DEFAULT_SENDER')
app.config['CACHE_TYPE'] = 'SimpleCache'
app.config['CACHE_DEFAULT_TIMEOUT'] = 300
app.config['JWT_SECRET_KEY'] = os.getenv('JWT_SECRET_KEY', 'jwt-secret-key-here')
app.config['JWT_ACCESS_TOKEN_EXPIRES'] = int(os.getenv('JWT_ACCESS_TOKEN_EXPIRES', 3600))
# Initialize extensions
db = SQLAlchemy(app)
mail = Mail(app)
bcrypt = Bcrypt(app)
jwt = JWTManager(app)
migrate = Migrate(app, db)
limiter = Limiter(
app=app,
key_func=get_remote_address,
default_limits=["200 per day", "50 per hour"]
)
cache = Cache(app)
api = Api(app)
swagger = Swagger(app)
# Configure logging
if not os.path.exists('logs'):
os.makedirs('logs')
handler = RotatingFileHandler('logs/app.log', maxBytes=10000, backupCount=3)
handler.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
app.logger.addHandler(handler)
# Create uploads directory if it doesn't exist
if not os.path.exists(app.config['UPLOAD_FOLDER']):
os.makedirs(app.config['UPLOAD_FOLDER'])
# Database models
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)
first_name = db.Column(db.String(50))
last_name = db.Column(db.String(50))
role = db.Column(db.String(20), default='user')
is_active = db.Column(db.Boolean, default=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
def set_password(self, password):
self.password_hash = bcrypt.generate_password_hash(password).decode('utf-8')
def check_password(self, password):
return bcrypt.check_password_hash(self.password_hash, password)
class Product(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(100), nullable=False)
description = db.Column(db.Text)
price = db.Column(db.Float, nullable=False)
stock = db.Column(db.Integer, default=0)
category_id = db.Column(db.Integer, db.ForeignKey('category.id'))
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
images = db.relationship('ProductImage', backref='product', lazy=True)
customizations = db.relationship('ProductCustomization', backref='product', lazy=True)
class Category(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(50), nullable=False)
description = db.Column(db.Text)
parent_id = db.Column(db.Integer, db.ForeignKey('category.id'))
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
products = db.relationship('Product', backref='category', lazy=True)
class ProductImage(db.Model):
id = db.Column(db.Integer, primary_key=True)
product_id = db.Column(db.Integer, db.ForeignKey('product.id'), nullable=False)
image_path = db.Column(db.String(255), nullable=False)
is_primary = db.Column(db.Boolean, default=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
class ProductCustomization(db.Model):
id = db.Column(db.Integer, primary_key=True)
product_id = db.Column(db.Integer, db.ForeignKey('product.id'), nullable=False)
customization_type = db.Column(db.String(50), nullable=False)
options = db.Column(db.JSON, nullable=False)
price_addition = db.Column(db.Float, default=0.0)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
class Order(db.Model):
id = db.Column(db.Integer, primary_key=True)
user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
status = db.Column(db.String(20), default='pending')
total_amount = db.Column(db.Float, nullable=False)
shipping_address = db.Column(db.Text, nullable=False)
billing_address = db.Column(db.Text, nullable=False)
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)
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)
product_id = db.Column(db.Integer, db.ForeignKey('product.id'), nullable=False)
quantity = db.Column(db.Integer, default=1)
customization_data = db.Column(db.JSON)
unit_price = db.Column(db.Float, nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
class Review(db.Model):
id = db.Column(db.Integer, primary_key=True)
product_id = db.Column(db.Integer, db.ForeignKey('product.id'), nullable=False)
user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
rating = db.Column(db.Integer, nullable=False)
comment = 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)
class Notification(db.Model):
id = db.Column(db.Integer, primary_key=True)
user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
type = db.Column(db.String(50), nullable=False)
message = db.Column(db.Text, nullable=False)
is_read = db.Column(db.Boolean, default=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
# Helper functions
def generate_unique_filename(filename):
ext = os.path.splitext(filename)[1]
unique_id = str(uuid.uuid4())[:8]
return f"{unique_id}{ext}"
def send_email(subject, recipients, html_body, text_body=None):
msg = Message(subject, recipients=recipients)
msg.body = text_body
msg.html = html_body
try:
mail.send(msg)
return True
except Exception as e:
app.logger.error(f"Failed to send email: {str(e)}")
return False
def validate_email(email):
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
return re.match(pattern, email) is not None
def validate_password(password):
if len(password) < 8:
return False
if not re.search(r'[A-Z]', password):
return False
if not re.search(r'[a-z]', password):
return False
if not re.search(r'[0-9]', password):
return False
if not re.search(r'[!@#$%^&*(),.?":{}|<>]', password):
return False
return True
def create_access_token(user_id):
return create_access_token(identity=user_id)
# API Resources
class UserResource(Resource):
@jwt_required()
def get(self):
current_user_id = get_jwt_identity()
user = User.query.get(current_user_id)
if not user:
return {'message': 'User not found'}, 404
return {
'id': user.id,
'username': user.username,
'email': user.email,
'first_name': user.first_name,
'last_name': user.last_name,
'role': user.role,
'is_active': user.is_active,
'created_at': user.created_at.isoformat(),
'updated_at': user.updated_at.isoformat()
}
@jwt_required()
def put(self):
current_user_id = get_jwt_identity()
data = request.get_json()
user = User.query.get(current_user_id)
if not user:
return {'message': 'User not found'}, 404
if 'password' in data:
if not validate_password(data['password']):
return {'message': 'Invalid password format'}, 400
user.set_password(data['password'])
if 'username' in data:
user.username = data['username']
if 'email' in data:
if not validate_email(data['email']):
return {'message': 'Invalid email format'}, 400
user.email = data['email']
if 'first_name' in data:
user.first_name = data['first_name']
if 'last_name' in data:
user.last_name = data['last_name']
db.session.commit()
return {
'message': 'User updated successfully',
'user': {
'id': user.id,
'username': user.username,
'email': user.email,
'first_name': user.first_name,
'last_name': user.last_name,
'role': user.role,
'is_active': user.is_active,
'created_at': user.created_at.isoformat(),
'updated_at': user.updated_at.isoformat()
}
}
class LoginResource(Resource):
def post(self):
data = request.get_json()
email = data.get('email')
password = data.get('password')
if not email or not password:
return {'message': 'Email and password are required'}, 400
user = User.query.filter_by(email=email).first()
if not user or not user.check_password(password):
return {'message': 'Invalid email or password'}, 401
access_token = create_access_token(identity=user.id)
return {
'access_token': access_token,
'token_type': 'bearer',
'user': {
'id': user.id,
'username': user.username,
'email': user.email,
'first_name': user.first_name,
'last_name': user.last_name,
'role': user.role,
'is_active': user.is_active
}
}
class RegisterResource(Resource):
def post(self):
data = request.get_json()
if not data or not all(key in data for key in ['username', 'email', 'password']):
return {'message': 'Username, email and password are required'}, 400
if not validate_email(data['email']):
return {'message': 'Invalid email format'}, 400
if not validate_password(data['password']):
return {'message': 'Password must be at least 8 characters long and contain at least one uppercase letter, one lowercase letter, one number and one special character'}, 400
if User.query.filter_by(username=data['username']).first():
return {'message': 'Username already exists'}, 409
if User.query.filter_by(email=data['email']).first():
return {'message': 'Email already exists'}, 409
user = User(
username=data['username'],
email=data['email'],
first_name=data.get('first_name', ''),
last_name=data.get('last_name', ''),
role='user'
)
user.set_password(data['password'])
try:
db.session.add(user)
db.session.commit()
# Send welcome email
welcome_subject = "Welcome to Our Platform"
welcome_body = f"""
Welcome, {data['username']}!
Thank you for registering with our platform.
Your account has been successfully created.
"""
send_email(welcome_subject, [data['email']], welcome_body)
return {
'message': 'User registered successfully',
'user': {
'id': user.id,
'username': user.username,
'email': user.email,
'first_name': user.first_name,
'last_name': user.last_name,
'role': user.role
}
}
except SQLAlchemyError as e:
db.session.rollback()
app.logger.error(f"Registration failed: {str(e)}")
return {'message': 'Registration failed'}, 500
# Add resources to API
api.add_resource(UserResource, '/api/users/me')
api.add_resource(LoginResource, '/api/auth/login')
api.add_resource(RegisterResource, '/api/auth/register')
# Routes
@app.route('/')
def index():
return render_template('index.html')
@app.route('/api/products')
def get_products():
products = Product.query.all()
return jsonify([{
'id': product.id,
'name': product.name,
'description': product.description,
'price': product.price,
'stock': product.stock,
'category': product.category.name if product.category else None,
'created_at': product.created_at.isoformat(),
'updated_at': product.updated_at.isoformat(),
'images': [image.image_path for image in product.images]
} for product in products])
@app.route('/api/products/
')
def get_product(product_id):
product = Product.query.get(product_id)
if not product:
return {'message': 'Product not found'}, 404
return {
'id': product.id,
'name': product.name,
'description': product.description,
'price': product.price,
'stock': product.stock,
'category': product.category.name if product.category else None,
'created_at': product.created_at.isoformat(),
'updated_at': product.updated_at.isoformat(),
'images': [image.image_path for image in product.images],
'customizations': [{
'id': customization.id,
'type': customization.customization_type,
'options': customization.options,
'price_addition': customization.price_addition
} for customization in product.customizations]
}
@app.route('/api/upload', methods=['POST'])
@jwt_required()
@limiter.limit("5 per minute")
def upload_file():
if 'file' not in request.files:
return {'message': 'No file part'}, 400
file = request.files['file']
if file.filename == '':
return {'message': 'No selected file'}, 400
if file:
filename = generate_unique_filename(file.filename)
filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename)
file.save(filepath)
return {
'message': 'File uploaded successfully',
'filename': filename,
'filepath': filepath
}
@app.route('/api/health')
def health_check():
return jsonify({'status': 'healthy'})
# Error handlers
@app.errorhandler(404)
def not_found(error):
return jsonify({'message': 'Resource not found'}), 404
@app.errorhandler(500)
def internal_error(error):
return jsonify({'message': 'Internal server error'}), 500
# JWT error handlers
@jwt.expired_token_loader
def expired_token_callback(expired_token):
return jsonify({
'message': 'The token has expired',
'expired_token': expired_token
}), 401
@jwt.invalid_token_loader
def invalid_token_callback(error):
return jsonify({
'message': 'Invalid token',
'error': str(error)
}), 401
@jwt.unauthorized_loader
def unauthorized_callback(error):
return jsonify({
'message': 'Authorization required',
'error': str(error)
}), 401
# Cache control
@app.after_request
def after_request(response):
cache_control = 'public, max-age=0'
if request.path.startswith('/api'):
cache_control = 'private, max-age=0'
response.headers['Cache-Control'] = cache_control
return response
if __name__ == '__main__':
app.run(debug=True)
