Ejercicio: Agenda de Contactos con Python y AWS ​

# app/__init__.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_marshmallow import Marshmallow
from config import config

db = SQLAlchemy()
migrate = Migrate()
ma = Marshmallow()

def create_app(config_name):
    app = Flask(__name__)
    
    # Cargar configuración
    app.config.from_object(config[config_name])
    config[config_name].init_app(app)
    
    # Inicializar extensiones
    db.init_app(app)
    migrate.init_app(app, db)
    ma.init_app(app)
    
    # Registrar blueprints
    from app.api import contacts_bp, auth_bp
    app.register_blueprint(contacts_bp, url_prefix='/api/contacts')
    app.register_blueprint(auth_bp, url_prefix='/api/auth')
    
    return app

# app/models/contact.py
from app import db
from datetime import datetime

class Contact(db.Model):
    __tablename__ = 'contacts'
    
    id = db.Column(db.Integer, primary_key=True)
    first_name = db.Column(db.String(50), nullable=False)
    last_name = db.Column(db.String(50), nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    phone = db.Column(db.String(20))
    address = db.Column(db.String(200))
    notes = db.Column(db.Text)
    user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    
    def __repr__(self):
        return f'<Contact {self.first_name} {self.last_name}>'

# app/schemas/contact.py
from app import ma
from app.models.contact import Contact
from marshmallow import fields, validates, ValidationError

class ContactSchema(ma.SQLAlchemySchema):
    class Meta:
        model = Contact

    id = ma.auto_field()
    first_name = fields.Str(required=True)
    last_name = fields.Str(required=True)
    email = fields.Email(required=True)
    phone = fields.Str()
    address = fields.Str()
    notes = fields.Str()
    created_at = fields.DateTime(dump_only=True)
    updated_at = fields.DateTime(dump_only=True)
    
    @validates('phone')
    def validate_phone(self, value):
        if value and not value.replace('+', '').isdigit():
            raise ValidationError('Invalid phone number format')

contact_schema = ContactSchema()
contacts_schema = ContactSchema(many=True)

# config/base.py
import os
from datetime import timedelta

class Config:
    # Configuración base
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'hard-to-guess-string'
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    
    # JWT
    JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY') or 'jwt-secret-string'
    JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=1)
    
    # Configuración de AWS
    AWS_REGION = os.environ.get('AWS_REGION', 'us-east-1')
    
    @staticmethod
    def init_app(app):
        pass

# config/dev.py
from .base import Config

class DevelopmentConfig(Config):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL')

# app/api/contacts.py
from flask import Blueprint, request, jsonify
from app.models.contact import Contact
from app.schemas.contact import contact_schema, contacts_schema
from app.services.contact_service import ContactService
from app.api.auth import jwt_required

contacts_bp = Blueprint('contacts', __name__)
contact_service = ContactService()

@contacts_bp.route('', methods=['GET'])
@jwt_required
def get_contacts():
    user_id = request.user.id
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 10, type=int)
    
    contacts = contact_service.get_contacts(user_id, page, per_page)
    return jsonify(contacts_schema.dump(contacts)), 200

@contacts_bp.route('/<int:contact_id>', methods=['GET'])
@jwt_required
def get_contact(contact_id):
    user_id = request.user.id
    contact = contact_service.get_contact(contact_id, user_id)
    if not contact:
        return jsonify({'message': 'Contact not found'}), 404
    return jsonify(contact_schema.dump(contact)), 200

@contacts_bp.route('', methods=['POST'])
@jwt_required
def create_contact():
    user_id = request.user.id
    data = request.get_json()
    
    errors = contact_schema.validate(data)
    if errors:
        return jsonify({'errors': errors}), 400
        
    contact = contact_service.create_contact(data, user_id)
    return jsonify(contact_schema.dump(contact)), 201

@contacts_bp.route('/<int:contact_id>', methods=['PUT'])
@jwt_required
def update_contact(contact_id):
    user_id = request.user.id
    data = request.get_json()
    
    errors = contact_schema.validate(data)
    if errors:
        return jsonify({'errors': errors}), 400
        
    contact = contact_service.update_contact(contact_id, data, user_id)
    if not contact:
        return jsonify({'message': 'Contact not found'}), 404
    return jsonify(contact_schema.dump(contact)), 200

@contacts_bp.route('/<int:contact_id>', methods=['DELETE'])
@jwt_required
def delete_contact(contact_id):
    user_id = request.user.id
    if contact_service.delete_contact(contact_id, user_id):
        return '', 204
    return jsonify({'message': 'Contact not found'}), 404

# app/services/contact_service.py
from app.models.contact import Contact
from app import db
from sqlalchemy.exc import SQLAlchemyError

class ContactService:
    def get_contacts(self, user_id, page=1, per_page=10):
        """Obtener lista paginada de contactos"""
        return Contact.query.filter_by(user_id=user_id)\
            .order_by(Contact.last_name)\
            .paginate(page=page, per_page=per_page, error_out=False)
    
    def get_contact(self, contact_id, user_id):
        """Obtener un contacto específico"""
        return Contact.query.filter_by(id=contact_id, user_id=user_id).first()
    
    def create_contact(self, data, user_id):
        """Crear un nuevo contacto"""
        try:
            contact = Contact(user_id=user_id, **data)
            db.session.add(contact)
            db.session.commit()
            return contact
        except SQLAlchemyError as e:
            db.session.rollback()
            raise e
    
    def update_contact(self, contact_id, data, user_id):
        """Actualizar un contacto existente"""
        contact = self.get_contact(contact_id, user_id)
        if not contact:
            return None
            
        try:
            for key, value in data.items():
                setattr(contact, key, value)
            db.session.commit()
            return contact
        except SQLAlchemyError as e:
            db.session.rollback()
            raise e
    
    def delete_contact(self, contact_id, user_id):
        """Eliminar un contacto"""
        contact = self.get_contact(contact_id, user_id)
        if not contact:
            return False
            
        try:
            db.session.delete(contact)
            db.session.commit()
            return True
        except SQLAlchemyError as e:
            db.session.rollback()
            raise e

# app/services/notification_service.py
import boto3
from botocore.exceptions import ClientError

class NotificationService:
    def __init__(self):
        self.sns = boto3.client('sns')
        self.topic_arn = os.environ.get('SNS_TOPIC_ARN')
    
    def notify_contact_created(self, contact):
        """Notificar cuando se crea un nuevo contacto"""
        try:
            message = {
                'action': 'contact_created',
                'contact_id': contact.id,
                'name': f'{contact.first_name} {contact.last_name}',
                'email': contact.email
            }
            
            self.sns.publish(
                TopicArn=self.topic_arn,
                Message=json.dumps(message),
                Subject='Nuevo Contacto Creado'
            )
        except ClientError as e:
            print(f"Error sending notification: {e}")
            
    def notify_contact_deleted(self, contact):
        """Notificar cuando se elimina un contacto"""
        try:
            message = {
                'action': 'contact_deleted',
                'contact_id': contact.id,
                'name': f'{contact.first_name} {contact.last_name}'
            }
            
            self.sns.publish(
                TopicArn=self.topic_arn,
                Message=json.dumps(message),
                Subject='Contacto Eliminado'
            )
        except ClientError as e:
            print(f"Error sending notification: {e}")

# app/api/errors.py
from flask import jsonify
from app import db
from sqlalchemy.exc import SQLAlchemyError

def register_error_handlers(app):
    @app.errorhandler(404)
    def not_found_error(error):
        return jsonify({'error': 'Resource not found'}), 404
        
    @app.errorhandler(400)
    def bad_request_error(error):
        return jsonify({'error': 'Bad request'}), 400
        
    @app.errorhandler(SQLAlchemyError)
    def database_error(error):
        db.session.rollback()
        return jsonify({'error': 'Database error occurred'}), 500
        
    @app.errorhandler(Exception)
    def internal_error(error):
        db.session.rollback()
        return jsonify({'error': 'Internal server error'}), 500

# tests/conftest.py
import pytest
from app import create_app, db
from app.models.contact import Contact
from app.models.user import User

@pytest.fixture
def app():
    app = create_app('testing')
    
    with app.app_context():
        db.create_all()
        yield app
        db.session.remove()
        db.drop_all()

@pytest.fixture
def client(app):
    return app.test_client()

@pytest.fixture
def test_user(app):
    with app.app_context():
        user = User(
            email='test@example.com',
            password='testpass123'
        )
        db.session.add(user)
        db.session.commit()
        return user

@pytest.fixture
def test_contact(app, test_user):
    with app.app_context():
        contact = Contact(
            first_name='John',
            last_name='Doe',
            email='john@example.com',
            phone='1234567890',
            user_id=test_user.id
        )
        db.session.add(contact)
        db.session.commit()
        return contact

# tests/test_api.py
def test_get_contacts(client, test_user, test_contact):
    token = create_access_token(test_user)
    headers = {'Authorization': f'Bearer {token}'}
    
    response = client.get('/api/contacts', headers=headers)
    assert response.status_code == 200
    data = response.get_json()
    assert len(data) == 1
    assert data[0]['email'] == test_contact.email

def test_create_contact(client, test_user):
    token = create_access_token(test_user)
    headers = {
        'Authorization': f'Bearer {token}',
        'Content-Type': 'application/json'
    }
    
    contact_data = {
        'first_name': 'Jane',
        'last_name': 'Doe',
        'email': 'jane@example.com',
        'phone': '0987654321'
    }
    
    response = client.post('/api/contacts', 
                         json=contact_data,
                         headers=headers)
    assert response.status_code == 201
    data = response.get_json()
    assert data['email'] == contact_data['email']