Compare commits

...

10 commits

14 changed files with 549 additions and 161 deletions

66
app.py
View file

@ -1,18 +1,62 @@
from flask import Flask
from models import db
from flask_migrate import Migrate
from routes import api
from routes.userRoutes import userRoutes
from routes.eventRoutes import eventRoutes
import config
from celery_utils import make_celery
from flask_jwt_extended import JWTManager
from middlewares.errorHandlers import handle_auth_error, handle_invalid_token
from flask_jwt_extended.exceptions import NoAuthorizationError
from jwt.exceptions import InvalidTokenError
from datetime import timedelta
app = Flask(__name__)
app.config.from_object(config.Config)
db.init_app(app)
migrate = Migrate(app, db)
celery = make_celery(app)
app.register_blueprint(api)
class App:
def __init__(self):
self.app = Flask(__name__)
self.set_config()
self.set_up_db()
self.set_up_jwt()
self.register_blueprints()
self.setup_error_handlers()
def set_config(self):
self.app.config.from_object(config.Config)
def set_up_db(self):
db.init_app(self.app)
self.migrate = Migrate(self.app, db)
def setup_error_handlers(self):
self.app.register_error_handler(NoAuthorizationError, handle_auth_error)
self.app.register_error_handler(InvalidTokenError, handle_invalid_token)
def set_up_jwt(self):
self.jwt_manager = JWTManager(self.app)
self.app.config['JWT_TOKEN_LOCATION'] = ['cookies']
self.app.config['JWT_COOKIE_NAME'] = 'access_token_cookie'
self.app.config['JWT_COOKIE_CSRF_PROTECT'] = False
self.app.config['JWT_ACCESS_TOKEN_EXPIRES'] = timedelta(days=30)
def register_blueprints(self):
self.app.register_blueprint(userRoutes, url_prefix='/user')
self.app.register_blueprint(eventRoutes, url_prefix='/event')
def run(self):
with self.app.app_context():
db.create_all()
self.app.run(debug=True)
def print_endpoints(self):
print("Endpoints and their functions:")
for rule in self.app.url_map.iter_rules():
print(f"Endpoint: {rule.endpoint}, Path: {rule}")
function_name = self.app.view_functions[rule.endpoint].__name__
print(f" Function: {function_name}")
app_class_instance = App()
app_instance = app_class_instance.app
if __name__ == '__main__':
with app.app_context():
db.create_all()
app.run(debug=True)
app_class_instance.print_endpoints()
app_class_instance.run()

View file

@ -1,2 +1,3 @@
class Config:
SQLALCHEMY_DATABASE_URI = 'sqlite:///events.db'
SQLALCHEMY_DATABASE_URI = 'sqlite:///events.db'
JWT_SECRET_KEY = 'your_jwt_secret_key'

View file

@ -0,0 +1,11 @@
from functools import wraps
from flask import request, jsonify
def handle_auth_error(e):
return jsonify({"error": "Please login"}), 401
def handle_invalid_token(e):
return jsonify({"error": "Invalid session token"}), 401
def handle_invalid_header_error(e):
return jsonify({"error": "Please login again"}), 401

View file

@ -1,6 +1,7 @@
from functools import wraps
from flask import request, jsonify
from datetime import datetime
from flask_jwt_extended import jwt_required, get_jwt_identity
def validate_event_post_request(f):
@wraps(f)
@ -21,10 +22,11 @@ def validate_event_post_request(f):
# Validate 'description'
if not isinstance(data['description'], str):
return jsonify({"message": "Invalid description"}), 400
# Validate 'time' (ensure it's a valid datetime string)
try:
datetime.strptime(data['duedate'], '%Y-%m-%dT%H:%M:%S')
time = datetime.strptime(data['duedate'], '%Y-%m-%dT%H:%M:%S')
if(time < datetime.now()):
return jsonify({"message": "Invalid time"}), 400
except ValueError:
return jsonify({"message": "Invalid time format. Use YYYY-MM-DDTHH:MM:SS"}), 400

View file

@ -0,0 +1,73 @@
from functools import wraps
from flask import request, jsonify, g
from flask_jwt_extended import jwt_required, get_jwt_identity
def validate_user_post_request(f):
@wraps(f)
def decorated_function(*args, **kwargs):
data = request.get_json()
if not data:
return jsonify({"message": "No input data provided"}), 400
# Check required fields
required_fields = ['username', 'password', 'email', 'location']
if not all(field in data for field in required_fields):
return jsonify({"message": "Please check your data, you missing some props; visit our docs https://git.dayanhub.com/kfir"}), 400
# Validate 'username'
if not isinstance(data['username'], str) or not data['username'].strip():
return jsonify({"message": "Invalid username"}), 400
# Validate 'password'
if not isinstance(data['password'], str) or not data['password'].strip():
return jsonify({"message": "Invalid password"}), 400
# Validate 'email'
if not isinstance(data['email'], str) or not data['email'].strip():
return jsonify({"message": "Invalid email"}), 400
# Validate 'location'
if not isinstance(data['location'], str) or not data['location'].strip():
return jsonify({"message": "Invalid location"}), 400
return f(*args, **kwargs)
return decorated_function
def validate_user_login_request(f):
@wraps(f)
def decorated_function(*args, **kwargs):
data = request.get_json()
if not data:
return jsonify({"message": "No input data provided"}), 400
# Check required fields
required_fields = ['email', 'password']
if not all(field in data for field in required_fields):
return jsonify({"message": "Please check your data, you missing some props; visit our docs https://git.dayanhub.com/kfir"}), 400
# Validate 'email'
if not isinstance(data['email'], str) or not data['email'].strip():
return jsonify({"message": "Invalid email"}), 400
# Validate 'password'
if not isinstance(data['password'], str) or not data['password'].strip():
return jsonify({"message": "Invalid password"}), 400
return f(*args, **kwargs)
return decorated_function
def authenticate_user(f):
@wraps(f)
@jwt_required(locations=["cookies"]) # Specify to look for the token in cookies
def decorated_function(*args, **kwargs):
# Get user identity from JWT
user_id = get_jwt_identity()
if user_id:
g.user_id = user_id
else:
return jsonify({"error": "Invalid session token"}), 401
return f(*args, **kwargs)
return decorated_function

View file

@ -1,42 +0,0 @@
"""initial migration
Revision ID: 5569d39a87cf
Revises:
Create Date: 2024-01-04 13:44:40.811421
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = '5569d39a87cf'
down_revision = None
branch_labels = None
depends_on = None
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
with op.batch_alter_table('event', schema=None) as batch_op:
batch_op.add_column(sa.Column('title', sa.String(length=100), nullable=False))
batch_op.add_column(sa.Column('description', sa.String(length=200), nullable=True))
batch_op.add_column(sa.Column('location', sa.String(length=100), nullable=True))
batch_op.add_column(sa.Column('deleted', sa.Boolean(), nullable=True))
batch_op.add_column(sa.Column('duedate', sa.DateTime(), nullable=True))
batch_op.add_column(sa.Column('created_at', sa.DateTime(), nullable=True))
# ### end Alembic commands ###
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
with op.batch_alter_table('event', schema=None) as batch_op:
batch_op.drop_column('created_at')
batch_op.drop_column('duedate')
batch_op.drop_column('deleted')
batch_op.drop_column('location')
batch_op.drop_column('description')
batch_op.drop_column('title')
# ### end Alembic commands ###

View file

@ -0,0 +1,49 @@
"""initial migration
Revision ID: bb3a88007475
Revises:
Create Date: 2024-01-07 14:12:47.164418
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = 'bb3a88007475'
down_revision = None
branch_labels = None
depends_on = None
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('user',
sa.Column('id', sa.String(length=36), nullable=False),
sa.Column('name', sa.String(length=100), nullable=True),
sa.Column('email', sa.String(length=120), nullable=False),
sa.Column('password_hash', sa.String(length=128), nullable=True),
sa.Column('location', sa.String(length=100), nullable=False),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('email')
)
op.create_table('event',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('title', sa.String(length=100), nullable=False),
sa.Column('description', sa.String(length=200), nullable=True),
sa.Column('location', sa.String(length=100), nullable=True),
sa.Column('deleted', sa.Boolean(), nullable=True),
sa.Column('duedate', sa.DateTime(), nullable=True),
sa.Column('created_at', sa.DateTime(), nullable=True),
sa.Column('user_id', sa.String(length=36), nullable=False),
sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
sa.PrimaryKeyConstraint('id')
)
# ### end Alembic commands ###
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table('event')
op.drop_table('user')
# ### end Alembic commands ###

View file

@ -1,6 +1,16 @@
from flask_sqlalchemy import SQLAlchemy
from flask_bcrypt import Bcrypt
from sqlalchemy import Table, Column, Integer, ForeignKey, String
import uuid
db = SQLAlchemy()
bcrypt = Bcrypt()
user_event_association = Table('user_event', db.Model.metadata,
Column('user_id', String(36), ForeignKey('user.id')),
Column('event_id', Integer, ForeignKey('event.id'))
)
class Event(db.Model):
id = db.Column(db.Integer, primary_key=True)
@ -10,6 +20,9 @@ class Event(db.Model):
deleted = db.Column(db.Boolean, default=False)
duedate = db.Column(db.DateTime)
created_at = db.Column(db.DateTime, default=db.func.now())
user_id = db.Column(db.String(36), db.ForeignKey('user.id'), nullable=False)
users = db.relationship('User', secondary=user_event_association, back_populates='events')
def to_dict(self):
return {
@ -18,5 +31,31 @@ class Event(db.Model):
'description': self.description,
'location': self.location,
'duedate': self.duedate.isoformat() if self.duedate else None,
'created_at': self.created_at.isoformat()
'created_at': self.created_at.isoformat(),
'user_id': self.user_id
}
class User(db.Model):
id = db.Column(db.String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
name = db.Column(db.String(100))
email = db.Column(db.String(120), unique=True, nullable=False)
password_hash = db.Column(db.String(128))
location = db.Column(db.String(100), nullable=False)
events = db.relationship('Event', secondary=user_event_association, back_populates='users')
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)
def to_dict(self):
return {
'id': self.id,
'name': self.name,
'email': self.email,
'location': self.location
}

View file

@ -1,61 +0,0 @@
from flask import Blueprint, jsonify, request
from services import EventService
from middlewares import validate_event_post_request
api = Blueprint('api', __name__)
# Create new event
@api.route('/events', methods=['POST'])
@validate_event_post_request
def create_event():
try:
data = request.json
new_event = EventService.create_event(data)
if new_event:
return jsonify(new_event.to_dict()), 201
else:
return jsonify({'error': 'Failed to create event'}), 400
except Exception as e:
return jsonify({'error': str(e)}), 500
# Get All Events
@api.route('/events', methods=['GET'])
def get_events():
try:
return {"events": EventService.get_all_events()}, 200
except Exception as e:
return {"error": str(e)}, 500
# Get Event by ID
@api.route('/events/<int:event_id>', methods=['GET'])
def get_event(event_id):
try:
return {"event": EventService.get_event_by_id(event_id)}, 200
except Exception as e:
return {"error": str(e)}, 500
# Update Event
@api.route('/events/<int:event_id>', methods=['PUT'])
@validate_event_post_request
def update_event(event_id):
try:
data = request.json
updated_event = EventService.update_event(event_id, data)
if updated_event:
return jsonify(updated_event.to_dict()), 200
else:
return jsonify({'error': 'Failed to update event'}), 400
except Exception as e:
return jsonify({'error': str(e)}), 500
# DELETE Event
@api.route('/events/<int:event_id>', methods=['DELETE'])
def delete_event(event_id):
try:
deleted_event = EventService.delete_event(event_id)
if deleted_event:
return jsonify(deleted_event.to_dict()), 200
else:
return jsonify({'error': 'Failed to delete event'}), 400
except Exception as e:
return jsonify({'error': str(e)}), 500

94
routes/eventRoutes.py Normal file
View file

@ -0,0 +1,94 @@
from flask import Blueprint, jsonify, request, g
from services.EventService import EventService
from middlewares.userMiddlewares import authenticate_user
from middlewares.eventMiddlewares import validate_event_post_request
eventRoutes = Blueprint('eventRoutes', __name__)
# Create new event
@eventRoutes.route('/', methods=['POST'])
@eventRoutes.route('', methods=['POST'])
@validate_event_post_request
@authenticate_user
def create_event():
try:
data = request.json
data['user_id'] = g.user_id
new_event = EventService.create_event(data)
if new_event:
return jsonify(new_event.to_dict()), 201
else:
return jsonify({'error': 'Failed to create event'}), 400
except Exception as e:
return jsonify({'error': str(e)}), 500
# Get All Events
@eventRoutes.route('/', methods=['GET'])
@eventRoutes.route('', methods=['GET'])
@authenticate_user
def get_events():
try:
user_events = EventService.get_upcoming_events(g.user_id)
return {"events": user_events, "count": len(user_events)}, 200
except Exception as e:
return {"error": str(e)}, 500
# Get Event by ID
@eventRoutes.route('/<int:event_id>', methods=['GET'])
@authenticate_user
def get_event(event_id):
try:
return {"event": EventService.get_event_by_id(event_id, g.user_id)}, 200
except Exception as e:
return {"error": str(e)}, 500
# Update Event
@eventRoutes.route('/<int:event_id>', methods=['PUT'])
@validate_event_post_request
@authenticate_user
def update_event(event_id):
try:
data = request.json
updated_event = EventService.update_event(event_id, data)
if updated_event:
return jsonify(updated_event.to_dict()), 200
else:
return jsonify({'error': 'Event not found'}), 404
except Exception as e:
return jsonify({'error': str(e)}), 500
# DELETE Event
@eventRoutes.route('/<int:event_id>', methods=['DELETE'])
@authenticate_user
def delete_event(event_id):
try:
deleted_event = EventService.delete_event(event_id)
if deleted_event:
return jsonify(deleted_event.to_dict()), 200
else:
return jsonify({'error': 'Event not found'}), 404
except Exception as e:
return jsonify({'error': str(e)}), 500
# User atended to event with id event_id and user must be loggedin
@eventRoutes.route('/<int:event_id>/attend', methods=['POST'])
@authenticate_user
def attend_event(event_id):
response = EventService.attend_event(event_id)
# Check if there's an error in the response
if 'error' in response:
return jsonify({'error': response['error']}), 400 # Or another appropriate status code
# If no error, return success response
return jsonify({'message': response['message']}), 200
@eventRoutes.route('/<int:event_id>/unattend', methods=['POST'])
@authenticate_user
def unattend_event(event_id):
response = EventService.unattend_event(event_id)
# Check if there's an error in the response
if 'error' in response:
return jsonify({'error': response['error']}), 400 # Or another appropriate status code
return jsonify({'message': response['message']}), 200

61
routes/userRoutes.py Normal file
View file

@ -0,0 +1,61 @@
from flask import Blueprint, jsonify, request
from services.UserService import UserService
from flask_jwt_extended import JWTManager, jwt_required, create_access_token
from middlewares.userMiddlewares import validate_user_post_request, validate_user_login_request
userRoutes = Blueprint('userRoutes', __name__)
@userRoutes.route('/', methods=['GET'])
@userRoutes.route('', methods=['GET'])
def allUsers():
users = UserService.get_all_users()
return jsonify(users), 200
@userRoutes.route('/', methods=['POST'])
@userRoutes.route('', methods=['POST'])
@validate_user_post_request
def createNewUser():
try:
data = request.json
if UserService.get_user_by_email(data['email']):
return jsonify({'error': 'User already exists'}), 400
new_user = UserService.create_user(data)
if new_user:
return jsonify(new_user), 201
else:
return jsonify({'error': 'Failed to create user'}), 400
except Exception as e:
return jsonify({'error': str(e)}), 500
@userRoutes.route('/login', methods=['POST'])
@validate_user_login_request
def loginUser():
try:
data = request.json
user = UserService.verify_user(data)
if user:
token = login(user)
response = user.to_dict()
response['token'] = token
response = jsonify(response)
return response, 200
else:
return jsonify({'error': 'Invalid credentials'}), 400
except Exception as e:
return jsonify({'error': str(e)}), 500
@userRoutes.route('/logout', methods=['POST'])
@jwt_required(optional=True)
def logoutUser():
try:
response = jsonify({'message': 'Logged out successfully'})
response.set_cookie('access_token_cookie', '', expires=0)
return response, 200
except Exception as e:
return jsonify({'error': str(e)}), 500
def login(user):
sessionToken = create_access_token(identity=user.id)
response = jsonify(user.to_dict())
response.set_cookie('access_token_cookie', sessionToken, httponly=True, path='/')
return sessionToken

View file

@ -1,43 +0,0 @@
from models import db, Event
from datetime import datetime
class EventService:
@staticmethod
def create_event(data):
new_event = Event(
title=data['title'],
description=data.get('description', ''),
location=data.get('location', ''),
duedate=datetime.strptime(data['duedate'], '%Y-%m-%dT%H:%M:%S')
)
db.session.add(new_event)
db.session.commit()
return new_event
@staticmethod
def get_all_events():
events=Event.query.filter_by(deleted=False).all()
return [event.to_dict() for event in events]
@staticmethod
def get_event_by_id(event_id):
return Event.query.filter_by(id=event_id, deleted=False).first().to_dict()
@staticmethod
def update_event(event_id, data):
event = Event.query.get(event_id)
event.title = data['title']
event.description = data.get('description', '')
event.location = data.get('location', '')
event.time = datetime.strptime(data['duedate'], '%Y-%m-%dT%H:%M:%S')
db.session.commit()
return event
@staticmethod
def delete_event(event_id):
event = Event.query.filter_by(id=event_id, deleted=False).first()
if event:
event.deleted = True
db.session.commit()
else:
return None
return event

123
services/EventService.py Normal file
View file

@ -0,0 +1,123 @@
from sqlalchemy.sql import exists
from models import db, Event, user_event_association
from services.UserService import UserService
from datetime import datetime
from flask import g
class EventService:
@staticmethod
def create_event(data):
new_event = Event(
title=data['title'],
description=data.get('description', ''),
location=data.get('location', ''),
duedate=datetime.strptime(data['duedate'], '%Y-%m-%dT%H:%M:%S'),
user_id=data['user_id']
)
db.session.add(new_event)
db.session.commit()
return new_event
@staticmethod
def get_upcoming_events(user_id):
events=Event.query.filter_by(deleted=False).all()
if(events):
return [event.to_dict() for event in events]
else:
return []
@staticmethod
def get_event_by_id(event_id, user_id):
event = Event.query.filter_by(id=event_id,user_id=user_id, deleted=False).first()
if(event):
return event.to_dict()
else:
return []
@staticmethod
def update_event(event_id, data):
event = Event.query.get(event_id)
if not event:
return None
event.title = data['title']
event.description = data.get('description', '')
event.location = data.get('location', '')
event.time = datetime.strptime(data['duedate'], '%Y-%m-%dT%H:%M:%S')
db.session.commit()
return event
@staticmethod
def delete_event(event_id):
event = Event.query.filter_by(id=event_id, deleted=False).first()
if event:
event.deleted = True
db.session.commit()
else:
return None
return event
@staticmethod
def attend_event(event_id):
user_id = g.user_id # Assuming user_id is stored in Flask's global g object
# Check if the event is valid and in the future
event = Event.query.filter(
Event.id == event_id,
Event.duedate > datetime.now(),
Event.deleted == False
).first()
if not event:
return {'error': 'Event not found or already passed'}
# Check if the user is already associated with the event
is_already_attending = db.session.query(exists().where(
user_event_association.c.user_id == user_id,
user_event_association.c.event_id == event_id
)).scalar()
if is_already_attending:
return {'error': 'User already attending this event'}
# Add the user to the event
user = UserService.get_user_by_id(user_id)
if not user:
return {'error': 'User not found'}
event.users.append(user)
db.session.commit()
return {'message': 'User successfully added to the event'}
@staticmethod
def unattend_event(event_id):
user_id = g.user_id # Assuming user_id is stored in Flask's global g object
# Check if the event is valid and in the future
event = Event.query.filter(
Event.id == event_id,
Event.duedate > datetime.now(),
Event.deleted == False
).first()
if not event:
return {'error': 'Event not found or already passed'}
# Check if the user is already associated with the event
is_already_attending = db.session.query(exists().where(
user_event_association.c.user_id == user_id,
user_event_association.c.event_id == event_id
)).scalar()
if not is_already_attending:
return {'error': 'User not attending this event'}
# Remove the user from the event
user = UserService.get_user_by_id(user_id)
if not user:
return {'error': 'User not found'}
event.users.remove(user)
db.session.commit()
return {'message': 'User successfully removed from the event'}

37
services/UserService.py Normal file
View file

@ -0,0 +1,37 @@
from flask_bcrypt import Bcrypt
from models import db, User
bcrypt = Bcrypt()
class UserService:
@staticmethod
def create_user(data):
new_user = User(
name=data['username'],
email=data['email'],
location=data['location'],
password_hash=bcrypt.generate_password_hash(data['password']).decode('utf-8')
)
db.session.add(new_user)
db.session.commit()
return new_user.to_dict()
@staticmethod
def get_all_users():
users = User.query.all()
return [user.to_dict() for user in users]
@staticmethod
def get_user_by_email(email):
return User.query.filter_by(email=email).first()
@staticmethod
def get_user_by_id(user_id):
return User.query.get(user_id)
@staticmethod
def verify_user(data):
user = UserService.get_user_by_email(data['email'])
if user and bcrypt.check_password_hash(user.password_hash, data['password']):
return user
return None