Untitled

Kool_Cool

Aug 12th, 2024

Never

Add comment

Not a member of Pastebin yet? Sign Up, it unlocks many cool features!

text 9.15 KB | None | 0 0

raw download clone embed print report

To convert your Django-based code into a FastAPI application, you need to follow these steps:
1. **Define Data Models**: Define the Pydantic models for request and response data.
2. **Set Up Database Connection**: Use an ORM like SQLAlchemy for database operations.
3. **Create API Endpoints**: Implement FastAPI endpoints to handle requests.
4. **Handle Data Processing**: Use appropriate libraries and methods to process and return the data.
Here’s how you can convert the given Django code to FastAPI:
### Step 1: Define Data Models
You'll need to define Pydantic models for request validation and response serialization.
```python
from pydantic import BaseModel
from typing import Optional, List
class ReportRequest(BaseModel):
created_date_min: str
created_date_max: str
trade_security__trade_exchange: str
trade_security__client: str
class ReportResponse(BaseModel):
trade_date: str
trade_security_no: str
expiry_date: Optional[str]
instrument_type: Optional[str]
symbol: Optional[str]
exchange_id: Optional[str]
trade_security_type_enum_id: Optional[str]
trade_tr: Optional[str]
trade_sr: Optional[str]
trade_no: Optional[str]
buy_sell_opt: Optional[str]
order_time: Optional[str]
quantity: Optional[int]
rate: Optional[float]
brokerage_total: Optional[float]
net_rate: Optional[float]
amount: Optional[float]
turnover: Optional[float]
stt: Optional[float]
sebi: Optional[float]
service_tax: Optional[float]
stamp_duty: Optional[float]
clear_charge: Optional[float]
```
### Step 2: Set Up Database Connection
You can use SQLAlchemy for database operations. Define the database models and create a session:
```python
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, Date, DateTime, Numeric
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from fastapi import FastAPI, HTTPException
DATABASE_URL = "postgresql://user:password@localhost/dbname"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
class TradeDetail(Base):
__tablename__ = "trade_detail"
id = Column(Integer, primary_key=True, index=True)
client_id = Column(String)
trade_exchange_id = Column(String)
# Define other fields and relationships
class TradeDetailItem(Base):
__tablename__ = "trade_detail_item"
id = Column(Integer, primary_key=True, index=True)
trade_security_id = Column(Integer, ForeignKey("trade_detail.id"))
trade_sr = Column(String)
trade_tr = Column(String)
# Define other fields and relationships
trade_security = relationship("TradeDetail")
```
### Step 3: Create FastAPI Endpoints
Define FastAPI endpoints that interact with your database and return data.
```python
from fastapi import FastAPI, Depends, Query
from sqlalchemy.orm import Session
from datetime import datetime
app = FastAPI()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.post("/generate-report", response_model=List[ReportResponse])
def generate_report(request: ReportRequest, db: Session = Depends(get_db)):
Today = datetime.today().date()
fin_year_start = get_financial_year_date() # Implement this function
date_first = request.created_date_min
date_second = request.created_date_max
date_first_obj = datetime.strptime(date_first, "%Y-%m-%d").date()
date_second_obj = datetime.strptime(date_second, "%Y-%m-%d").date()
exchange = request.trade_security__trade_exchange
client_id = request.trade_security__client
# Convert dates as needed
if date_first:
date_begin = datetime.strptime(date_first, "%Y-%m-%d").strftime('%Y-%m-%d')
date_end = datetime.strptime(date_second, "%Y-%m-%d").strftime('%Y-%m-%d')
else:
date_begin = datetime.strptime(fin_year_start, "%d.%m.%Y").strftime('%Y-%m-%d')
date_end = datetime.strptime(Today, "%d.%m.%Y").strftime('%Y-%m-%d')
# Query database
trdMainObj = db.query(TradeDetailItem).join(TradeDetail).filter(
TradeDetail.client_id == client_id,
TradeDetail.trade_exchange_id == exchange,
TradeDetailItem.trade_date.between(date_begin, date_end)
).all()
if not trdMainObj:
raise HTTPException(status_code=404, detail="No records found")
return trdMainObj
```
### Step 4: Handle Data Processing
If you need to process the data or convert it to a specific format, handle it in the endpoint or in a utility function.
```python
def convert_to_dict(data):
# Convert SQLAlchemy results to dict
return [{col.name: getattr(row, col.name) for col in row.__table__.columns} for row in data]
```
### Complete Example
Here’s a complete example integrating all parts:
```python
from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy.orm import Session
from pydantic import BaseModel
from datetime import datetime
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, Date, DateTime, Numeric
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from typing import List, Optional
# Database Setup
DATABASE_URL = "postgresql://user:password@localhost/dbname"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
class TradeDetail(Base):
__tablename__ = "trade_detail"
id = Column(Integer, primary_key=True, index=True)
client_id = Column(String)
trade_exchange_id = Column(String)
# Define other fields
class TradeDetailItem(Base):
__tablename__ = "trade_detail_item"
id = Column(Integer, primary_key=True, index=True)
trade_security_id = Column(Integer, ForeignKey("trade_detail.id"))
trade_sr = Column(String)
trade_tr = Column(String)
trade_date = Column(Date)
# Define other fields
trade_security = relationship("TradeDetail")
Base.metadata.create_all(bind=engine)
# Pydantic Models
class ReportRequest(BaseModel):
created_date_min: str
created_date_max: str
trade_security__trade_exchange: str
trade_security__client: str
class ReportResponse(BaseModel):
trade_date: str
trade_security_no: str
expiry_date: Optional[str]
instrument_type: Optional[str]
symbol: Optional[str]
exchange_id: Optional[str]
trade_security_type_enum_id: Optional[str]
trade_tr: Optional[str]
trade_sr: Optional[str]
trade_no: Optional[str]
buy_sell_opt: Optional[str]
order_time: Optional[str]
quantity: Optional[int]
rate: Optional[float]
brokerage_total: Optional[float]
net_rate: Optional[float]
amount: Optional[float]
turnover: Optional[float]
stt: Optional[float]
sebi: Optional[float]
service_tax: Optional[float]
stamp_duty: Optional[float]
clear_charge: Optional[float]
# FastAPI App
app = FastAPI()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.post("/generate-report", response_model=List[ReportResponse])
def generate_report(request: ReportRequest, db: Session = Depends(get_db)):
Today = datetime.today().date()
fin_year_start = get_financial_year_date() # Implement this function
date_first = request.created_date_min
date_second = request.created_date_max
date_first_obj = datetime.strptime(date_first, "%Y-%m-%d").date()
date_second_obj = datetime.strptime(date_second, "%Y-%m-%d").date()
exchange = request.trade_security__trade_exchange
client_id = request.trade_security__client
# Convert dates as needed
if date_first:
date_begin = datetime.strptime(date_first, "%Y-%m-%d").strftime('%Y-%m-%d')
date_end = datetime.strptime(date_second, "%Y-%m-%d").strftime('%Y-%m-%d')
else:
date_begin = datetime.strptime(fin_year_start, "%d.%m.%Y").strftime('%Y-%m-%d')
date_end = datetime.strptime(Today, "%d.%m.%Y").strftime('%Y-%m-%d')
# Query database
trdMainObj = db.query(TradeDetailItem).join(TradeDetail).filter(
TradeDetail.client_id == client_id,
TradeDetail.trade_exchange_id == exchange,
TradeDetailItem.trade_date.between(date_begin, date_end)
).all()
if not trdMainObj:
raise HTTPException(status_code=404, detail="No records found")
return trdMainObj
```
### Summary
- **Data Models**: Define models with Pydantic for request validation and response serialization.
- **Database Connection**: Use SQLAlchemy for database operations.
- **API Endpoints**: Create FastAPI endpoints to handle requests
and return responses.
- **Data Processing**: Convert SQLAlchemy results to dictionaries as needed.
This FastAPI application should give you a good starting point for migrating from Django to FastAPI. Adjust the code as necessary based on your specific requirements and database schema.

Add Comment

Please, Sign In to add comment