login register

app/__init__.py

@@ -10,6 +10,7 @@ from .config import Config
 from flask_mail import Mail
 from itsdangerous import URLSafeTimedSerializer
 from flasgger import Swagger
+from datetime import datetime
 
 db = SQLAlchemy()
 migrate = Migrate()
@@ -101,4 +102,12 @@ def create_app():
         from app.seeder import seed_users
         seed_users()
 
+    @jwt.expired_token_loader
+    def expired_token_callback(jwt_header, jwt_payload):
+        return jsonify({
+            "status": 401,
+            "message": "The token has expired",
+            "expired_at": datetime.fromtimestamp(jwt_payload['exp']).isoformat()
+        }), 401
+
     return app

app/config.py

@@ -12,7 +12,7 @@ class Config:
     DEBUG = os.getenv('FLASK_ENV') == 'development'
     PORT = int(os.getenv('PORT', 8000))
     JWT_SECRET_KEY = os.getenv('JWT_SECRET_KEY', 'myjwtsecretkey')
-    JWT_ACCESS_TOKEN_EXPIRES = datetime.timedelta(minutes=15)
+    JWT_ACCESS_TOKEN_EXPIRES = datetime.timedelta(minutes=1440)
     JWT_REFRESH_TOKEN_EXPIRES = datetime.timedelta(days=7)
     MAIL_SERVER = os.getenv('MAIL_SERVER', 'smtp.example.com')
     MAIL_PORT = os.getenv('MAIL_PORT', 587)

app/helper/Master.py

@@ -19,7 +19,6 @@ from Crypto.Cipher import AES
 import json
 import sys
 
-
 # PHRASE        = 'AygqrawSf1GI2FFzUX8AnRjLxb0V1paOuwMR22UR_1I='
 PHRASE_KY   = b'0123456789abcdef' # Key should be 16 bytes (128 bits)
 PHRASE_IV   = b'abcdef9876543210'   # IV should be 16 bytes
@@ -266,3 +265,10 @@ class Master(object):
         # iv_b64 = base64.b64encode(iv).decode()
 
         return encrypted_data_b64
+
+    def row2dict(row):
+        d = {}
+        for column in row.__table__.columns:
+            d[column.name] = str(getattr(row, column.name))
+
+        return d

app/models/user.py

 from app import db
 from werkzeug.security import generate_password_hash, check_password_hash
 from app.config import Config
+from datetime import datetime
+import pytz
+
+jakarta_timezone = pytz.timezone('Asia/Jakarta')
+current_time = datetime.now(jakarta_timezone)
     
 class User(db.Model):
-    __tablename__ = 'user'
+    __tablename__ = 'auth_user'
     __table_args__ = {'schema': Config.SCHEMA_NAME }
     
     id = db.Column(db.Integer, primary_key=True)
+    created_date = db.Column(db.DateTime, nullable=False)
+    updated_date = db.Column(db.DateTime, nullable=False)
+    telegram_id = db.Column(db.Integer, nullable=False)
+    role_id = db.Column(db.Integer, nullable=False)
+    password = db.Column(db.String(500), nullable=False)
+    file_path = db.Column(db.String(500), nullable=True)
     username = db.Column(db.String(150), unique=True, nullable=False)
-    password_hash = db.Column(db.String(255), nullable=False)
-    email = db.Column(db.String(150), unique=True, nullable=True, default='no-reply@example.com')
+    first_name = db.Column(db.String(150), unique=True, nullable=False)
+    last_name = db.Column(db.String(150), unique=True, nullable=False)
+    email = db.Column(db.String(150), unique=True, nullable=True, default='admin@admin.com')
     current_refresh_token = db.Column(db.String(500), nullable=True)
+
     # roles = db.relationship('Role', secondary='user_roles', backref=db.backref('users', lazy='dynamic'))
 
+
     def set_password(self, password):
-        self.password_hash = generate_password_hash(password)
+        self.password = generate_password_hash(password)
 
     def check_password(self, password):
-        return check_password_hash(self.password_hash, password)
+        return check_password_hash(self.password, password)
 
 # Role model
-class Role(db.Model):
-    __tablename__ = 'role'
-    __table_args__ = {'schema': Config.SCHEMA_NAME }
+# class Role(db.Model):
+#     __tablename__ = 'role'
+#     __table_args__ = {'schema': Config.SCHEMA_NAME }
 
-    id = db.Column(db.Integer, primary_key=True)
-    name = db.Column(db.String(50), unique=True)
+#     id = db.Column(db.Integer, primary_key=True)
+#     name = db.Column(db.String(50), unique=True)
     # privileges = db.relationship('Privilege', secondary='role_privileges', backref=db.backref('roles', lazy='dynamic'))
 
 # Privilege model
-class Privilege(db.Model):
+# class Privilege(db.Model):
 
-    __tablename__ = 'privilege'
-    __table_args__ = {'schema': Config.SCHEMA_NAME }
+#     __tablename__ = 'privilege'
+#     __table_args__ = {'schema': Config.SCHEMA_NAME }
 
-    id = db.Column(db.Integer, primary_key=True)
-    name = db.Column(db.String(50), unique=True)
+#     id = db.Column(db.Integer, primary_key=True)
+#     name = db.Column(db.String(50), unique=True)
 
 # # Association tables
 # user_roles = db.Table('user_roles',

app/routes/auth.py

@@ -4,9 +4,15 @@ from app.models.user import User
 from app.utils.response import success_response, error_response
 from app import db, jwt
 import datetime
+from datetime import datetime as dtm
 from flask_mail import Message
 from itsdangerous import URLSafeTimedSerializer
 from flasgger import swag_from
+import json
+import pytz
+
+jakarta_timezone = pytz.timezone('Asia/Jakarta')
+current_time = dtm.now(jakarta_timezone)
 
 auth_bp = Blueprint('auth', __name__)
 
@@ -30,14 +36,18 @@ auth_bp = Blueprint('auth', __name__)
 })
 def register():
     data = request.get_json()
+    created_date = current_time
+    first_name = data.get('first_name')
+    last_name = data.get('last_name')
     username = data.get('username')
     password = data.get('password')
     email   = data.get('email')
+    role_id = data.get('role')
     
     if User.query.filter_by(username=username).first():
         return error_response(message="User already exists", code=400)
 
-    new_user = User(username=username, email=email)
+    new_user = User(username=username, email=email, created_date=created_date, first_name=first_name, last_name=last_name, role_id=role_id)
     new_user.set_password(password)
     db.session.add(new_user)
     db.session.commit()
@@ -70,8 +80,20 @@ def login():
     # Update user with the new refresh token
     user.current_refresh_token = refresh_token
     db.session.commit()
+    usr = {
+            'id': user.id,
+            'username': user.username,
+            'created_date': user.created_date,
+            'updated_date': user.updated_date,
+            'role_id': user.role_id,
+            'first_name': user.first_name,
+            'last_name': user.last_name,
+            'email': user.email,
+            "access_token": access_token, 
+            "refresh_token": refresh_token
+        }
     
-    return success_response(data={"access_token": access_token, "refresh_token": refresh_token})
+    return success_response(data=usr)
 
 @auth_bp.route('/logout', methods=['GET'])
 @jwt_required(refresh=True)

app/routes/user.py

@@ -3,6 +3,7 @@ from flask_jwt_extended import jwt_required, get_jwt_identity
 from app.models.user import User
 from app.utils.response import success_response, error_response
 from app import db, jwt
+from app.helper.Master import Master
 
 user_bp = Blueprint('user', __name__)
 
@@ -11,11 +12,14 @@ user_bp = Blueprint('user', __name__)
 def profile():
     current_user = get_jwt_identity()  # This should be a string (username)
     user = User.query.filter_by(username=current_user).first()
+    
     if user.current_refresh_token is None:
         return error_response("User is not Loggin", code=403)
 
     if user:
-        return success_response(data={"username": user.username, "email": user.email})
+        if 'password' in Master.row2dict(user):
+            del Master.row2dict(user)['password']
+        return success_response(data= Master.row2dict(user))
 
     return error_response("User not found", code=404)
 
@@ -23,14 +27,14 @@ def profile():
 @jwt_required()
 def get_users():
     users = User.query.all()
-    users_data = [{"id": user.id, "username": user.username, "email": user.email, "roles": [role.name for role in user.roles]} for user in users]
+    users_data = [{"id": user.id, "username": user.username, "email": user.email, "role_id": [role.name for role in user.roles]} for user in users]
     return success_response(data=users_data)
 
 @user_bp.route('/<int:user_id>', methods=['GET'])
 @jwt_required()
 def get_user(user_id):
     user = User.query.get_or_404(user_id)
-    user_data = {"id": user.id, "username": user.username, "email": user.email, "roles": [role.name for role in user.roles]}
+    user_data = {"id": user.id, "username": user.username, "email": user.email, "role_id": [role.name for role in user.roles]}
     return success_response(data=user_data)
 
 @user_bp.route('/create', methods=['POST'])

app/seeder/seed_users.py

 from app import db
 from app.models.user import User
+from datetime import datetime
+import pytz
+
+jakarta_timezone = pytz.timezone('Asia/Jakarta')
+current_time = datetime.now(jakarta_timezone)
 
 def seed_users():
-    if not User.query.filter_by(username='admin').first():
-        admin = User(username='admin')
-        admin.email = 'admin@example.com'
+    if not User.query.filter_by(username='superadmin').first():
+        admin = User(username='superadmin')
+        admin.created_date = current_time
+        admin.updated_date = current_time
+        admin.role_id = '0'
         admin.set_password('password')
+        admin.username = 'superadmin'
+        admin.first_name = 'super'
+        admin.last_name = 'admin'
+        admin.email = 'superadmin@admin.com'
+
         db.session.add(admin)
         db.session.commit()

requirements.txt

@@ -30,6 +30,7 @@ pycryptodome==3.20.0
 PyJWT==2.8.0
 python-dateutil==2.9.0.post0
 python-dotenv==1.0.1
+pytz==2024.1
 PyYAML==6.0.1
 referencing==0.35.1
 requests==2.32.3