salim4n's picture
Upload 5 files
c577353 verified
import pandas as pd
import sqlite3
import requests
from typing import List, Dict, Any
import os
from sqlalchemy import create_engine, Column, Float, String, Integer, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# Create base class for declarative models
Base = declarative_base()
class Freight(Base):
"""SQLAlchemy model for freight data"""
__tablename__ = "freights"
id = Column(Integer, primary_key=True)
departure = Column(DateTime)
origin_port_locode = Column(String)
origin_port_name = Column(String)
destination_port = Column(String)
destination_port_name = Column(String)
dv20rate = Column(Float)
dv40rate = Column(Float)
currency = Column(String)
inserted_on = Column(DateTime)
def download_csv(url: str, local_path: str = "freights.csv") -> str:
"""
Download CSV file from Hugging Face and save it locally
"""
response = requests.get(url)
with open(local_path, "wb") as f:
f.write(response.content)
return local_path
def create_database(db_name: str = "freights.db") -> None:
"""
Create SQLite database and necessary tables
"""
engine = create_engine(f"sqlite:///{db_name}")
Base.metadata.create_all(engine)
def load_csv_to_db(csv_path: str, db_name: str = "freights.db") -> None:
"""
Load CSV data into SQLite database
"""
# Read CSV
df = pd.read_csv(csv_path, parse_dates=["departure", "inserted_on"])
# Connect to database
engine = create_engine(f"sqlite:///{db_name}")
# Save to database
df.to_sql("freights", engine, if_exists="replace", index=False)
def initialize_database(csv_url: str) -> None:
"""
Initialize the database by downloading CSV and loading data.
Args:
csv_url: URL of the CSV file to download and load.
"""
# Download CSV
csv_path = download_csv(csv_url)
# Create and load database
create_database()
load_csv_to_db(csv_path)
print("Database initialized.")
# Clean up CSV file
if os.path.exists(csv_path):
os.remove(csv_path)