Python Forum
Thread Rating:
  • 0 Vote(s) - 0 Average
  • 1
  • 2
  • 3
  • 4
  • 5
SQLAlchemy ORM question
#1
  • I have written python software that finds and downloads all of the csv files at a particular ftp location.
  • I then programatically create an SQLAlchemy ORM Model (.py file) by examining the contents of each csv header, and data.
  • I then execute the Model code, which creates my database.
So good so far.
  • Now I would like to automatically populate each database table from the csv files, without referencing the column names of each model class column. Although I am confident that this can be done, I'm having difficulty formulating how to do so.

Please help me think this through. Remember, this is using the ORM model.

I will share complete code as a forum tutorial when finished.
Reply
#2
Progress:

First of all, I apologise for not posting more information last evening. I was tired and frustrated, so neglected to do so.
So, here's some required details.

To run this code:
  1. create a new project directory and cd to that diretory.
  2. run python -m venv venv to create a virtual environment.
  3. activate the virtual environment (Linux): . ./venv/bin/activate
  4. Add an empty __init__.py script to project directory
  5. create a directory named src and navigate to it.
  6. Add an empty __init__.py script
  7. Add the two scripts below (named as shown) to the src directory

ScsvlPaths.py
import os
from pathlib import Path


class ScsvlPaths:
    def __init__(self, depth=0):
        os.chdir(os.path.abspath(os.path.dirname(__file__)))
        dir_depth = abs(depth)

        HomePath = Path(".")

        while dir_depth:
            HScsvlPathsomePath = HomePath / ".."
            dir_depth -= 1

        rootpath = HomePath / ".."

        self.datapath = rootpath / "data"
        self.datapath.mkdir(exist_ok=True)

        self.csvpath = self.datapath / "csv"
        self.csvpath.mkdir(exist_ok=True)

        self.databasepath = self.datapath / "database"
        self.databasepath.mkdir(exist_ok=True)

        self.docpath = rootpath / "docs"
        self.docpath.mkdir(exist_ok=True)

        self.jsonpath = self.datapath / "json"
        self.jsonpath.mkdir(exist_ok=True)

        self.htmlpath = self.datapath / "html"
        self.htmlpath.mkdir(exist_ok=True)

        self.prettypath = self.datapath / "pretty"
        self.prettypath.mkdir(exist_ok=True)

        self.skeletons = self.datapath / 'skeletons'
        self.skeletons.mkdir(exist_ok=True)

        self.srcpath = rootpath / 'src'
        self.srcpath.mkdir(exist_ok=True)

        self.tmppath = self.datapath / "tmp"
        self.tmppath.mkdir(exist_ok=True)

        # Database
        self.FlightAidsDb = self.databasepath / "New.db"



if __name__ == '__main__':
    ScsvlPaths()
DbModel.py - This script was automatically generated from other code that I have written.
which I will publish if I get this all to work.
from ScsvlPaths import ScsvlPaths
from datetime import datetime
from sqlalchemy import Column, String, Integer, Date, \
    create_engine, DateTime, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, backref


fpath = ScsvlPaths()
engine = create_engine(f"sqlite:///{fpath.FlightAidsDb}")
Base = declarative_base()


class airport_frequencies(Base):

    __tablename__ = 'airport_frequencies'


    id = Column(Integer, primary_key=True)

    airport_ref = Column(Integer)
    airport_ident = Column(String(10))
    type = Column(String(24))
    description = Column(String(96))
    frequency_mhz = Column(String(10))

    def __init__(self,  id,  airport_ref,  airport_ident,  type,  description,  frequency_mhz,):
        self.id = id
        self.airport_ref = airport_ref
        self.airport_ident = airport_ident
        self.type = type
        self.description = description
        self.frequency_mhz = frequency_mhz


    def __repr__(self):
        return f"<airport_frequencies , id, airport_ref, airport_ident, type, description, frequency_mhz>"

class airports(Base):

    __tablename__ = 'airports'


    id = Column(Integer, primary_key=True)

    ident = Column(String(10))
    type = Column(String(21))
    name = Column(String(120))
    latitude_deg = Column(String(36))
    longitude_deg = Column(String(31))
    elevation_ft = Column(String(7))
    continent = Column(String(3))
    iso_country = Column(String(3))
    iso_region = Column(String(10))
    municipality = Column(String(72))
    scheduled_service = Column(String(4))
    gps_code = Column(String(6))
    iata_code = Column(String(4))
    local_code = Column(String(10))
    home_link = Column(String(192))
    wikipedia_link = Column(String(192))
    keywords = Column(String(384))

    def __init__(self,  id,  ident,  type,  name,  latitude_deg,  longitude_deg,  elevation_ft,  continent,  iso_country,  iso_region,  municipality,  scheduled_service,  gps_code,  iata_code,  local_code,  home_link,  wikipedia_link,  keywords,):
        self.id = id
        self.ident = ident
        self.type = type
        self.name = name
        self.latitude_deg = latitude_deg
        self.longitude_deg = longitude_deg
        self.elevation_ft = elevation_ft
        self.continent = continent
        self.iso_country = iso_country
        self.iso_region = iso_region
        self.municipality = municipality
        self.scheduled_service = scheduled_service
        self.gps_code = gps_code
        self.iata_code = iata_code
        self.local_code = local_code
        self.home_link = home_link
        self.wikipedia_link = wikipedia_link
        self.keywords = keywords


    def __repr__(self):
        return f"<airports , id, ident, type, name, latitude_deg, longitude_deg, elevation_ft, continent, iso_country, iso_region, municipality, scheduled_service, gps_code, iata_code, local_code, home_link, wikipedia_link, keywords>"

class countries(Base):

    __tablename__ = 'countries'


    id = Column(Integer, primary_key=True)

    code = Column(String(3))
    name = Column(String(66))
    continent = Column(String(3))
    wikipedia_link = Column(String(111))
    keywords = Column(String(76))

    def __init__(self,  id,  code,  name,  continent,  wikipedia_link,  keywords,):
        self.id = id
        self.code = code
        self.name = name
        self.continent = continent
        self.wikipedia_link = wikipedia_link
        self.keywords = keywords


    def __repr__(self):
        return f"<countries , id, code, name, continent, wikipedia_link, keywords>"

class navaids(Base):

    __tablename__ = 'navaids'


    id = Column(Integer, primary_key=True)

    filename = Column(String(60))
    ident = Column(String(12))
    name = Column(String(46))
    type = Column(String(10))
    frequency_khz = Column(String(9))
    latitude_deg = Column(String(31))
    longitude_deg = Column(String(31))
    elevation_ft = Column(String(7))
    iso_country = Column(String(3))
    dme_frequency_khz = Column(String(9))
    dme_channel = Column(String(6))
    dme_latitude_deg = Column(String(12))
    dme_longitude_deg = Column(String(13))
    dme_elevation_ft = Column(String(6))
    slaved_variation_deg = Column(String(12))
    magnetic_variation_deg = Column(String(12))
    usageType = Column(String(12))
    power = Column(String(10))
    associated_airport = Column(String(10))

    def __init__(self,  id,  filename,  ident,  name,  type,  frequency_khz,  latitude_deg,  longitude_deg,  elevation_ft,  iso_country,  dme_frequency_khz,  dme_channel,  dme_latitude_deg,  dme_longitude_deg,  dme_elevation_ft,  slaved_variation_deg,  magnetic_variation_deg,  usageType,  power,  associated_airport,):
        self.id = id
        self.filename = filename
        self.ident = ident
        self.name = name
        self.type = type
        self.frequency_khz = frequency_khz
        self.latitude_deg = latitude_deg
        self.longitude_deg = longitude_deg
        self.elevation_ft = elevation_ft
        self.iso_country = iso_country
        self.dme_frequency_khz = dme_frequency_khz
        self.dme_channel = dme_channel
        self.dme_latitude_deg = dme_latitude_deg
        self.dme_longitude_deg = dme_longitude_deg
        self.dme_elevation_ft = dme_elevation_ft
        self.slaved_variation_deg = slaved_variation_deg
        self.magnetic_variation_deg = magnetic_variation_deg
        self.usageType = usageType
        self.power = power
        self.associated_airport = associated_airport


    def __repr__(self):
        return f"<navaids , id, filename, ident, name, type, frequency_khz, latitude_deg, longitude_deg, elevation_ft, iso_country, dme_frequency_khz, dme_channel, dme_latitude_deg, dme_longitude_deg, dme_elevation_ft, slaved_variation_deg, magnetic_variation_deg, usageType, power, associated_airport>"

class regions(Base):

    __tablename__ = 'regions'


    id = Column(Integer, primary_key=True)

    code = Column(String(10))
    local_code = Column(String(6))
    name = Column(String(78))
    continent = Column(String(3))
    iso_country = Column(String(3))
    wikipedia_link = Column(String(126))
    keywords = Column(String(139))

    def __init__(self,  id,  code,  local_code,  name,  continent,  iso_country,  wikipedia_link,  keywords,):
        self.id = id
        self.code = code
        self.local_code = local_code
        self.name = name
        self.continent = continent
        self.iso_country = iso_country
        self.wikipedia_link = wikipedia_link
        self.keywords = keywords


    def __repr__(self):
        return f"<regions , id, code, local_code, name, continent, iso_country, wikipedia_link, keywords>"

class runways(Base):

    __tablename__ = 'runways'


    id = Column(Integer, primary_key=True)

    airport_ref = Column(Integer)
    airport_ident = Column(String(10))
    length_ft = Column(String(9))
    width_ft = Column(String(6))
    surface = Column(String(94))
    lighted = Column(Integer)
    closed = Column(Integer)
    le_ident = Column(String(10))
    le_latitude_deg = Column(String(13))
    le_longitude_deg = Column(String(13))
    le_elevation_ft = Column(String(7))
    le_heading_degT = Column(String(9))
    le_displaced_threshold_ft = Column(String(6))
    he_ident = Column(String(10))
    he_latitude_deg = Column(String(13))
    he_longitude_deg = Column(String(13))
    he_elevation_ft = Column(String(7))
    he_heading_degT = Column(String(9))
    he_displaced_threshold_ft = Column(String(6))

    def __init__(self,  id,  airport_ref,  airport_ident,  length_ft,  width_ft,  surface,  lighted,  closed,  le_ident,  le_latitude_deg,  le_longitude_deg,  le_elevation_ft,  le_heading_degT,  le_displaced_threshold_ft,  he_ident,  he_latitude_deg,  he_longitude_deg,  he_elevation_ft,  he_heading_degT,  he_displaced_threshold_ft,):
        self.id = id
        self.airport_ref = airport_ref
        self.airport_ident = airport_ident
        self.length_ft = length_ft
        self.width_ft = width_ft
        self.surface = surface
        self.lighted = lighted
        self.closed = closed
        self.le_ident = le_ident
        self.le_latitude_deg = le_latitude_deg
        self.le_longitude_deg = le_longitude_deg
        self.le_elevation_ft = le_elevation_ft
        self.le_heading_degT = le_heading_degT
        self.le_displaced_threshold_ft = le_displaced_threshold_ft
        self.he_ident = he_ident
        self.he_latitude_deg = he_latitude_deg
        self.he_longitude_deg = he_longitude_deg
        self.he_elevation_ft = he_elevation_ft
        self.he_heading_degT = he_heading_degT
        self.he_displaced_threshold_ft = he_displaced_threshold_ft


    def __repr__(self):---------------- table: regions ----------------
id: regions.id
code: regions.code
local_code: regions.local_code
name: regions.name
continent: regions.continent
iso_country: regions.iso_country
wikipedia_link: regions.wikipedia_link
keywords: regions.keywords
        return f"<runways , id, airport_ref, airport_ident, length_ft, width_ft, surface, lighted, closed, le_ident, le_latitude_deg, le_longitude_deg, le_elevation_ft, le_heading_degT, le_displaced_threshold_ft, he_ident, he_latitude_deg, he_longitude_deg, he_elevation_ft, he_heading_degT, he_displaced_threshold_ft>"

class DbModel:
    def __init__(self):
        Base.metadata.create_all(engine)


if __name__ == '__main__':
    DbModel()

  1. from command line (in src directory) run: python ScsvlPaths.py
    This will create the directory structure for the project.
  2. Some data is needed for running the code below, so
  3. Below is the program that I am working on.

Play.py
from ScsvlPaths import ScsvlPaths
import DbModel


class Play4:
    def __init__(self):
        self.spath = ScsvlPaths()
        self.model = DbModel

    def get_column_list(self, stem):
        columns = []
        tablename = stem.replace('-', '_')
        tabclass = getattr(self.model, tablename)
        cols = vars(tabclass)
        for key, value in cols.items():
            if not (key.startswith('_')):
                columns.append(key)
        return tablename, columns

    def play(self):
        filelist = [x for x in self.spath.csvpath.iterdir() if x.is_file() and x.suffix == '.csv']
        for filename in filelist:
            tablename = filename.stem.replace('-', '_')
            class_contents = getattr(self.model, tablename).__dict__
            print(f"\n---------------- table: {tablename} ----------------")
            for key, value in class_contents.items():
                if not key.startswith('_'):
                    print(f"{key}: {value}")

def main():
    p4 = Play4()
    p4.play()


if __name__ == '__main__':
    main()
output with two files downloaded above:
Output:
---------------- table: countries ---------------- id: countries.id code: countries.code name: countries.name continent: countries.continent wikipedia_link: countries.wikipedia_link keywords: countries.keywords ---------------- table: regions ---------------- id: regions.id code: regions.code local_code: regions.local_code name: regions.name continent: regions.continent iso_country: regions.iso_country wikipedia_link: regions.wikipedia_link keywords: regions.keywords
So I have the necessary model attributes needed to populate the table (i think).
There's probably an easier way to do this, but as the old saying goes, sometimes you can't see the forest because of the trees, and my mind is kind of giddy right now.

If anyone has the stamina to get this far, I thank you.
Any better ideas?
Reply
#3
I've figured it out, and as stated will be releasing as a tutorial soon.

The finished code will update old files, create an sqlalchemy ORM model, create the database, and populate same.
Reply


Forum Jump:

User Panel Messages

Announcements
Announcement #1 8/1/2020
Announcement #2 8/2/2020
Announcement #3 8/6/2020