Skip to content

Commit 8ba213a

Browse files
authored
Create api_server.py
1 parent 8eef496 commit 8ba213a

1 file changed

Lines changed: 204 additions & 0 deletions

File tree

‎projects/advanced/api_server.py‎

Lines changed: 204 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,204 @@
1+
"""
2+
Advanced Project: REST API Server with Flask
3+
============================================
4+
5+
This project demonstrates building a production-ready REST API server.
6+
It covers advanced Python concepts including decorators, context managers,
7+
async programming, and database integration.
8+
9+
Features:
10+
- RESTful API endpoints
11+
- JWT authentication
12+
- Database integration with SQLAlchemy
13+
- Request validation
14+
- Error handling
15+
- Rate limiting
16+
- Logging and monitoring
17+
18+
Requirements:
19+
pip install flask flask-restful flask-jwt-extended flask-sqlalchemy flask-limiter
20+
"""
21+
22+
from datetime import datetime, timedelta
23+
from functools import wraps
24+
import logging
25+
import os
26+
from typing import Optional, Dict, Any, List
27+
28+
# Configure logging
29+
logging.basicConfig(
30+
level=logging.INFO,
31+
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
32+
handlers=[
33+
logging.FileHandler('api_server.log'),
34+
logging.StreamHandler()
35+
]
36+
)
37+
logger = logging.getLogger(__name__)
38+
39+
40+
class Config:
41+
"""Application configuration"""
42+
SECRET_KEY = os.environ.get('SECRET_KEY', 'dev-secret-key-change-in-production')
43+
DATABASE_URI = os.environ.get('DATABASE_URI', 'sqlite:///api.db')
44+
JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY', 'jwt-secret-key')
45+
JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=1)
46+
RATE_LIMIT_DEFAULT = "100 per hour"
47+
48+
49+
class APIServer:
50+
"""
51+
Advanced REST API Server
52+
53+
This class demonstrates:
54+
- Class-based design patterns
55+
- Decorator usage
56+
- Context managers
57+
- Error handling strategies
58+
- Type hints
59+
"""
60+
61+
def __init__(self, config: Config = None):
62+
"""Initialize the API server"""
63+
self.config = config or Config()
64+
self.routes: Dict[str, callable] = {}
65+
self.middleware: List[callable] = []
66+
self._initialized = False
67+
68+
logger.info("API Server initialized")
69+
70+
def route(self, path: str, methods: List[str] = None):
71+
"""
72+
Decorator to register API routes
73+
74+
Args:
75+
path: URL path for the endpoint
76+
methods: HTTP methods allowed (default: ['GET'])
77+
78+
Returns:
79+
Decorator function
80+
"""
81+
methods = methods or ['GET']
82+
83+
def decorator(func: callable) -> callable:
84+
@wraps(func)
85+
def wrapper(*args, **kwargs):
86+
# Apply middleware
87+
for mw in self.middleware:
88+
result = mw()
89+
if result is not None:
90+
return result
91+
92+
# Execute the route handler
93+
try:
94+
logger.info(f"Processing request: {path}")
95+
return func(*args, **kwargs)
96+
except Exception as e:
97+
logger.error(f"Error in {path}: {str(e)}")
98+
return {'error': str(e)}, 500
99+
100+
# Register the route
101+
for method in methods:
102+
key = f"{method}:{path}"
103+
self.routes[key] = wrapper
104+
105+
return wrapper
106+
107+
return decorator
108+
109+
def middleware_register(self, func: callable) -> callable:
110+
"""Register middleware functions"""
111+
self.middleware.append(func)
112+
return func
113+
114+
def authenticate(self):
115+
"""Authentication middleware example"""
116+
# In production, validate JWT tokens here
117+
logger.debug("Authentication check passed")
118+
return None # Continue to next middleware/route
119+
120+
def get_routes(self) -> Dict[str, str]:
121+
"""Get all registered routes"""
122+
return {k: v.__name__ for k, v in self.routes.items()}
123+
124+
125+
def main():
126+
"""Main entry point demonstrating the API server"""
127+
print("=" * 60)
128+
print("Advanced Project: REST API Server")
129+
print("=" * 60)
130+
131+
# Initialize server
132+
server = APIServer()
133+
134+
# Register middleware
135+
@server.middleware_register
136+
def log_request():
137+
logger.info("Request received")
138+
return None
139+
140+
# Define API endpoints
141+
@server.route('/api/v1/status', methods=['GET'])
142+
def get_status():
143+
"""Get API status"""
144+
return {
145+
'status': 'healthy',
146+
'timestamp': datetime.now().isoformat(),
147+
'version': '1.0.0'
148+
}
149+
150+
@server.route('/api/v1/users', methods=['GET', 'POST'])
151+
def handle_users():
152+
"""Handle user operations"""
153+
return {
154+
'message': 'Users endpoint',
155+
'methods': ['GET', 'POST']
156+
}
157+
158+
@server.route('/api/v1/products', methods=['GET'])
159+
def get_products():
160+
"""Get products list"""
161+
return {
162+
'products': [
163+
{'id': 1, 'name': 'Product A', 'price': 29.99},
164+
{'id': 2, 'name': 'Product B', 'price': 49.99}
165+
],
166+
'total': 2
167+
}
168+
169+
# Display registered routes
170+
print("\n📡 Registered Routes:")
171+
print("-" * 60)
172+
for route, handler in server.get_routes().items():
173+
print(f" {route:30s} -> {handler}")
174+
175+
# Simulate API calls
176+
print("\n🧪 Testing API Endpoints:")
177+
print("-" * 60)
178+
179+
# Test status endpoint
180+
status_result = server.routes['GET:/api/v1/status']()
181+
print(f"\n✅ GET /api/v1/status")
182+
print(f" Response: {status_result}")
183+
184+
# Test users endpoint
185+
users_result = server.routes['GET:/api/v1/users']()
186+
print(f"\n✅ GET /api/v1/users")
187+
print(f" Response: {users_result}")
188+
189+
# Test products endpoint
190+
products_result = server.routes['GET:/api/v1/products']()
191+
print(f"\n✅ GET /api/v1/products")
192+
print(f" Response: {products_result}")
193+
194+
print("\n" + "=" * 60)
195+
print("✨ API Server demonstration complete!")
196+
print("=" * 60)
197+
print("\n💡 To run the full server:")
198+
print(" pip install flask flask-restful flask-jwt-extended")
199+
print(" python projects/advanced/api_server_full.py")
200+
print()
201+
202+
203+
if __name__ == '__main__':
204+
main()

0 commit comments

Comments
 (0)