100 lines
3.0 KiB
Python
100 lines
3.0 KiB
Python
import datetime
|
|
import psycopg as pg
|
|
from dotenv import load_dotenv
|
|
import os
|
|
|
|
from pydantic import BaseModel, Field
|
|
from typing import List, Optional
|
|
|
|
# Load environment variables from .env
|
|
load_dotenv()
|
|
|
|
# Fetch variables
|
|
USER = os.getenv("user")
|
|
PASSWORD = os.getenv("password")
|
|
HOST = os.getenv("host")
|
|
PORT = os.getenv("port")
|
|
DBNAME = os.getenv("database")
|
|
POOLMODE = os.getenv("pool_mode")
|
|
|
|
def get_connection():
|
|
# Connect to the database
|
|
try:
|
|
connection = pg.connect(
|
|
user=USER,
|
|
password=PASSWORD,
|
|
host=HOST,
|
|
port=PORT,
|
|
dbname=DBNAME,
|
|
# pool_mode=POOLMODE
|
|
)
|
|
print("Connection successful!")
|
|
|
|
except Exception as e:
|
|
print(f"Failed to connect: {e}")
|
|
|
|
return connection
|
|
|
|
|
|
def map_rows(model: type[BaseModel], cursor) -> list[BaseModel]:
|
|
columns = [desc[0] for desc in cursor.description]
|
|
return [model.model_validate(dict(zip(columns, row))) for row in cursor.fetchall()]
|
|
|
|
|
|
class Player(BaseModel):
|
|
id: Optional[int] = None # bigint, identity, primary key
|
|
created_at: Optional[datetime.datetime] = Field(default_factory=datetime.datetime.now) # optional, defaults to now
|
|
name: Optional[str] = None # text, not null
|
|
cost: Optional[int] = None # smallint, not null
|
|
image: Optional[str] = None # text, not null
|
|
image2: Optional[str] = None # text, nullable
|
|
cardset_id: Optional[int] = None # bigint, not null, foreign key
|
|
set_num: Optional[int] = None # integer, nullable
|
|
headshot: Optional[str] = None # text, nullable
|
|
vanity_card: Optional[str] = None # text, nullable
|
|
strat_code: Optional[str] = None # text, nullable
|
|
bbref_id: Optional[str] = None # text, nullable
|
|
fangr_id: Optional[str] = None # text, nullable
|
|
description: Optional[str] = None # text, not null
|
|
quantity: Optional[int] = Field(default=999) # smallint, not null, default 999
|
|
franchise: Optional[str] = None # enum or text? Adjust if you have a Franchise model or enum
|
|
mlbclub: Optional[str] = None # enum or text? Same as above
|
|
positions: Optional[List[str]] = Field(default_factory=lambda: ['DH']) # text[], nullable, default '{DH}'
|
|
rarity_id: Optional[int] = None # bigint, not null, foreign key
|
|
mlbplayer_id: Optional[int] = None # bigint, nullable, foreign key
|
|
|
|
|
|
def main():
|
|
connection = get_connection()
|
|
|
|
# Create a cursor to execute SQL queries
|
|
cursor = connection.cursor()
|
|
|
|
# Example query
|
|
# cursor.execute("SELECT NOW();")
|
|
# result = cursor.fetchone()
|
|
# print("Current Time:", result)
|
|
|
|
cursor.execute(
|
|
'''
|
|
SELECT id, name, cost, image, headshot
|
|
FROM players
|
|
WHERE cost > 950 and cost < 1000
|
|
'''
|
|
)
|
|
|
|
# Not working, yet. query is missing fields that were previously required
|
|
players = map_rows(Player, cursor)
|
|
|
|
# Close the cursor and connection
|
|
cursor.close()
|
|
connection.close()
|
|
print("Connection closed.")
|
|
|
|
for row in players:
|
|
print(row)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|