SrAsafe/Meu_projeto
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Mensagem aqui

Browse Source
This commit is contained in:
Antero Asafe Matos de Oliveira 2025-05-16 19:55:05 -04:00
parent 572a248721
commit 8aff1b9c4e
13 changed files with 264 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
.vscode/settings.json vendored Normal file
View File

@ -0,0 +1,5 @@
{
"python.analysis.extraPaths": [
"./app"
]
}

1
__init__.py
View File

@ -1 +0,0 @@
__init__.py

11
app.py
View File

@ -1,11 +0,0 @@
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
return {"mensagem": "API está funcionando"}
if __name__ == "__main__":
import uvicorn
uvicorn.run("app:app", host="127.0.0.1", port=5000, reload=True)

36
app/__init__.py Normal file
View File

@ -0,0 +1,36 @@
from flask import Flask
from app.database import db
from app.models.user import User
from app.routes.user_routes import user_bp
from werkzeug.security import generate_password_hash
def create_default_user():
# Verifica se o banco de dados está vazio
user = User.query.first()
if not user:
# Cria um usuário padrão com senha "senha123"
default_user = User(
name="admin",
email="admin@example.com",
password=generate_password_hash("senha123", method="pbkdf2:sha256")
)
# Adiciona o usuário no banco de dados
db.session.add(default_user)
db.session.commit()
print("Usuário padrão criado com sucesso!")
def create_app():
app = Flask(__name__)
app.config.from_object("config")
db.init_app(app)
# Registrar Blueprints
app.register_blueprint(user_bp, url_prefix="/users")
return app

31
app/auth.py Normal file
View File

@ -0,0 +1,31 @@
from functools import wraps
from flask import request, jsonify
import jwt
from app.models.user import User # ou Organization
SECRET_KEY = "sua_chave_secreta"
def generate_token(user):
token = jwt.encode({"id": user.id}, SECRET_KEY, algorithm="HS256")
return token
def token_required(f):
@wraps(f)
def decorated(*args, **kwargs):
token = None
if 'Authorization' in request.headers:
token = request.headers['Authorization'].split(" ")[1] # Bearer <token>
if not token:
return jsonify({"message": "Token ausente!"}), 401
try:
data = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
current_user = User.query.get(data["id"])
except:
return jsonify({"message": "Token inválido!"}), 401
return f(current_user, *args, **kwargs)
return decorated

29
app/database.py
View File

@ -1,28 +1,3 @@
from sqlalchemy import create_engine from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import Session
from .database import SessionLocal
db = SQLAlchemy()
# URL do banco - aqui usando SQLite local
SQLALCHEMY_DATABASE_URL = "sqlite:///./meubanco.db"
# Para SQLite, o argumento check_same_thread é necessário
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
# Sessão de conexão com o banco
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Base para os modelos
Base = declarative_base()
# Dependência para obter uma sessão do banco de dados
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()

22
app/models/documentos.py
View File

@ -1,19 +1,13 @@
from app.database import db
import uuid import uuid
from sqlalchemy import Column, String, DateTime, ForeignKey, Integer
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base() class Documento(db.Model):
class Documento(Base):
__tablename__ = 'documentos' __tablename__ = 'documentos'
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, unique=True, nullable=False) id = db.Column(db.UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, unique=True, nullable=False)
nome_arquivo = Column(String, nullable=False) nome_arquivo = db.Column(db.String, nullable=False)
caminho_arquivo = Column(String, nullable=False) # armazena o caminho do arquivo original caminho_arquivo = db.Column(db.String, nullable=False)
usuario_id = Column(Integer, ForeignKey('users.id'), nullable=False) usuario_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
criado_em = Column(DateTime(timezone=True), server_default=func.now(), nullable=False) criado_em = db.Column(db.DateTime(timezone=True), server_default=db.func.now(), nullable=False)
usuario = relationship('User', back_populates='documentos') usuario = db.relationship('User', back_populates='documentos')

18
app/models/organizacao.py
View File

@ -1,14 +1,10 @@
from sqlalchemy import Column, Integer, String, DateTime, func from app.database import db
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base() class Organizacao(db.Model):
__tablename__ = 'organizacao'
class Organizacao(Base): id = db.Column(db.Integer, primary_key=True, autoincrement=True, unique=True)
__tablename__ = 'organizacoes' nome = db.Column(db.String, nullable=False, unique=True)
criado_em = db.Column(db.DateTime(timezone=True), server_default=db.func.now(), nullable=False)
id = Column(Integer, primary_key=True, autoincrement=True, unique=True) usuarios = db.relationship('User', back_populates='organizacao')
nome = Column(String, nullable=False, unique=True)
criado_em = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
usuarios = relationship('User', back_populates='organizacao')

26
app/models/user.py
View File

@ -1,19 +1,15 @@
from sqlalchemy import Column, Integer, String, ForeignKey, DateTime, func from app.database import db
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base() class User(db.Model):
class User(Base):
__tablename__ = 'users' __tablename__ = 'users'
id = Column(Integer, primary_key=True, autoincrement=True, unique=True) id = db.Column(db.Integer, primary_key=True, autoincrement=True, unique=True)
nome = Column(String, nullable=False) nome = db.Column(db.String, nullable=False)
email = Column(String, nullable=False, unique=True) email = db.Column(db.String, nullable=False, unique=True)
senha_hash = Column(String, nullable=False) senha_hash = db.Column(db.String, nullable=False)
organizacao_id = Column(Integer, ForeignKey('organizacoes.id'), nullable=False) organizacao_id = db.Column(db.Integer, db.ForeignKey('organizacao.id'), nullable=False)
criado_em = Column(DateTime(timezone=True), server_default=func.now(), nullable=False) criado_em = db.Column(db.DateTime(timezone=True), server_default=db.func.now(), nullable=False)
atualizado_em = Column(DateTime(timezone=True), onupdate=func.now()) atualizado_em = db.Column(db.DateTime(timezone=True), onupdate=db.func.now())
organizacao = relationship('Organizacao', back_populates='usuarios') organizacao = db.relationship('Organizacao', back_populates='usuarios')
documentos = relationship('Documento', back_populates='usuario') documentos = db.relationship('Documento', back_populates='usuario')

84
app/routes/documentos_routes.py
View File

@ -1,31 +1,63 @@
from fastapi import APIRouter, HTTPException, Depends from flask import Blueprint, jsonify, request
from sqlalchemy.orm import Session from app.auth import token_required
from database import get_db from app.models.documentos import document
from app.database import db
router_documento = APIRouter(prefix="/documentos", tags=["Documentos"]) document_bp = Blueprint("documents", __name__)
# Listar todos os documentos
@document_bp.route("/", methods=["GET"])
@token_required
def get_documents(self):
documents = document.query.all()
return jsonify([
{"id": d.id, "title": d.title, "content": d.content, "owner_id": d.user_id}
for d in documents
])
@router_documento.post("/") # Criar um novo documento
def criar_documento( @document_bp.route("/", methods=["POST"])
nome_arquivo: str, @token_required
caminho_arquivo: str, def create_document(self):
usuario_id: int, data = request.get_json()
db: Session = Depends(get_db), title = data.get("title")
): content = data.get("content")
documento = documento(
nome_arquivo=nome_arquivo,
caminho_arquivo=caminho_arquivo,
usuario_id=usuario_id,
)
db.add(documento)
db.commit()
db.refresh(documento)
return documento
new_doc = document(title=title, content=content, user_id=self.id) # self.id vem do token
@router_documento.get("/{documento_id}") db.session.add(new_doc)
def obter_documento(documento_id: str, db: Session = Depends(get_db)): db.session.commit()
documento = db.query(documento).filter(documento.id == documento_id).first()
if not documento: return jsonify({"message": "Documento criado com sucesso!", "document_id": new_doc.id}), 201
raise HTTPException(status_code=404, detail="Documento não encontrado")
return documento # Obter um documento específico
@document_bp.route("/<int:document_id>", methods=["GET"])
@token_required
def get_document(self, document_id):
document = document.query.get_or_404(document_id)
return jsonify({
"id": document.id,
"title": document.title,
"content": document.content,
"owner_id": document.user_id
})
# Atualizar um documento
@document_bp.route("/<int:document_id>", methods=["PUT"])
@token_required
def update_document(self, document_id):
document = document.query.get_or_404(document_id)
data = request.get_json()
document.title = data.get("title", document.title)
document.content = data.get("content", document.content)
db.session.commit()
return jsonify({"message": "Documento atualizado com sucesso!"})
# Deletar um documento
@document_bp.route("/<int:document_id>", methods=["DELETE"])
@token_required
def delete_document(self, document_id):
document = document.query.get_or_404(document_id)
db.session.delete(document)

63
app/routes/organizacao_routes.py
View File

@ -1,25 +1,46 @@
from fastapi import APIRouter, HTTPException, Depends from flask import Blueprint, jsonify, request
from sqlalchemy.orm import Session from app.auth import generate_token, token_required
from database import get_db from app.models.organizacao import organization
from app.database import db
from werkzeug.security import generate_password_hash, check_password_hash
organization_bp = Blueprint("organizations", __name__)
router_organizacao = APIRouter(prefix="/organizacoes", tags=["Organizacoes"]) @organization_bp.route("/", methods=["GET"])
@token_required
def get_organizations(self):
organizations = organization.query.all()
return jsonify([{"id": o.id, "name": o.name, "email": o.email} for o in organizations])
@router_organizacao.post("/") @organization_bp.route("/", methods=["POST"])
def criar_organizacao(nome: str, db: Session = Depends(get_db)): def create_organization(self):
organizacao = db.query(organizacao).filter(organizacao.nome == nome).first() data = request.get_json()
if organizacao: name = data.get("name")
raise HTTPException(status_code=400, detail="Organização já existe") email = data.get("email")
password = data.get("password")
nova_organizacao = organizacao(nome=nome)
db.add(nova_organizacao)
db.commit()
db.refresh(nova_organizacao)
return nova_organizacao
@router_organizacao.get("/{organizacao_id}") if organization.query.filter_by(email=email).first():
def obter_organizacao(organizacao_id: int, db: Session = Depends(get_db)): return jsonify({"message": "Organização já existe!"}), 400
organizacao = db.query(organizacao).filter(organizacao.id == organizacao_id).first()
if not organizacao: hashed_password = generate_password_hash(password, method="pbkdf2:sha256")
raise HTTPException(status_code=404, detail="Organização não encontrada")
return organizacao new_org = organization(name=name, email=email, password=hashed_password)
db.session.add(new_org)
db.session.commit()
return jsonify({"message": "Organização registrada com sucesso!"}), 201
@organization_bp.route("/login", methods=["POST"])
def login_organization():
data = request.get_json()
email = data.get("email")
password = data.get("password")
organization = organization.query.filter_by(email=email).first()
if not organization or not check_password_hash(organization.password, password):
return jsonify({"message": "Credenciais inválidas!"}), 401
token = generate_token(organization)
return jsonify({"message": "Login bem-sucedido", "token": token}), 200

69
app/routes/user_routes.py
View File

@ -1,22 +1,55 @@
from fastapi import APIRouter, HTTPException, Depends from flask import Blueprint, jsonify, request
from sqlalchemy.orm import Session from app.auth import generate_token, token_required
from models import User # seus models from app.models.user import User
from database import get_db from app.database import db
from werkzeug.security import generate_password_hash, check_password_hash
user_bp = Blueprint("users", __name__)
router_user = APIRouter(prefix="/users", tags=["Users"]) @user_bp.route("/", methods=["GET"])
@token_required # Protege a rota
def get_users(self):
users = User.query.all() # select * from User
return jsonify([{"id": u.id, "name": u.name, "email": u.email} for u in users])
@router_user.post("/") @user_bp.route("/", methods=["POST"])
def criar_user(nome: str, email: str, senha_hash: str, organizacao_id: int, db: Session = Depends(get_db)): def create_user(self):
user = User(nome=nome, email=email, senha_hash=senha_hash, organizacao_id=organizacao_id) data = request.get_json()
db.add(user) print(data)
db.commit() nome = data.get("nome")
db.refresh(user) email = data.get("email")
return user senha = data.get("senha")
@router_user.get("/{user_id}") # Verificar se o usuário já existe
def obter_user(user_id: int, db: Session = Depends(get_db)): if User.query.filter_by(email=email).first():
user = db.query(User).filter(User.id == user_id).first() return jsonify({"message": "Usuário já existe!"}), 400
if not user:
raise HTTPException(status_code=404, detail="Usuário não encontrado") # Gerar senha criptografada
return user hashed_password = generate_password_hash(senha, method="pbkdf2:sha256")
# Criar novo usuário
new_user = User(nome=nome, email=email, senha=senha)
# Adicionar e salvar no banco de dados
db.session.add(new_user)
db.session.commit()
return jsonify({"message": "Usuário registrado com sucesso!"}), 201
# Rota para login de usuário
@user_bp.route("/login", methods=["POST"])
def login_user():
data = request.get_json()
print(data)
email = data.get("email")
senha = data.get("senha")
user = User.query.filter_by(email=email).first()
if not user or not check_password_hash(user.senha, senha):
return jsonify({"message": "Credenciais inválidas!"}), 401
token = generate_token(user)
return jsonify({"message": "Login bem-sucedido", "token": token}), 200

13
run.py Normal file
View File

@ -0,0 +1,13 @@
from app import create_default_user, create_app
from app.database import db
app = create_app()
if __name__ == "__main__":
with app.app_context():
db.create_all()
create_default_user()
print("Banco de dados criado")
app.run(debug=True, port=5000)