-
Notifications
You must be signed in to change notification settings - Fork 127
Expand file tree
/
Copy pathseed_database.py
More file actions
141 lines (112 loc) · 4.79 KB
/
seed_database.py
File metadata and controls
141 lines (112 loc) · 4.79 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
import csv
import os
import sys
import random
from datetime import datetime, timedelta
from collections import defaultdict
# Add the parent directory to sys.path to allow importing from models
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from flask import Flask
from models import init_db, db, Breed, Dog
from models.dog import AdoptionStatus
def create_app():
"""Create and configure Flask app for database operations"""
app = Flask(__name__)
# Get the server directory path (one level up from utils)
server_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
db_path = os.environ.get('DATABASE_PATH', os.path.join(server_dir, 'dogshelter.db'))
app.config['SQLALCHEMY_DATABASE_URI'] = f'sqlite:///{db_path}'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# Initialize the database with the app
init_db(app)
return app
def create_breeds():
"""Seed the database with breeds from the CSV file"""
app = create_app()
# Path to the CSV file
csv_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
'models', 'breeds.csv')
with app.app_context():
# Check if breeds already exist
existing_breeds = Breed.query.count()
if existing_breeds > 0:
print(f"Database already contains {existing_breeds} breeds. Skipping seed.")
return
# Read the CSV file and add breeds to the database
with open(csv_path, 'r') as file:
csv_reader = csv.DictReader(file)
for row in csv_reader:
breed = Breed(name=row['Breed'], description=row['Description'])
db.session.add(breed)
# Commit the changes
db.session.commit()
# Verify the seeding
breed_count = Breed.query.count()
print(f"Successfully seeded {breed_count} breeds to the database.")
def create_dogs():
"""Seed the database with dogs from the CSV file, ensuring at least 3 dogs per breed"""
app = create_app()
# Path to the CSV file
csv_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
'models', 'dogs.csv')
with app.app_context():
# Check if dogs already exist
existing_dogs = Dog.query.count()
if existing_dogs > 0:
print(f"Database already contains {existing_dogs} dogs. Skipping seed.")
return
# Get all breeds from the database
breeds = Breed.query.all()
if not breeds:
print("No breeds found in database. Please seed breeds first.")
return
# Track how many dogs are assigned to each breed
breed_counts = defaultdict(int)
# Read the CSV file
dogs_data = []
with open(csv_path, 'r') as file:
csv_reader = csv.DictReader(file)
for row in csv_reader:
dogs_data.append(row)
def create_dog(dog_info, breed_id):
"""Helper function to create a dog with consistent attributes"""
dog = Dog(
name=dog_info['Name'],
description=dog_info['Description'],
breed_id=breed_id,
age=int(dog_info['Age']),
gender=dog_info['Gender'],
status=random.choice(list(AdoptionStatus)),
intake_date=datetime.now() - timedelta(days=random.randint(1, 365))
)
db.session.add(dog)
breed_counts[breed_id] += 1
return dog
# First pass: assign at least 3 dogs to each breed
for breed in breeds:
# Get 3 random dogs that haven't been assigned yet
for _ in range(3):
if not dogs_data:
break
dog_info = random.choice(dogs_data)
dogs_data.remove(dog_info)
create_dog(dog_info, breed.id)
# Second pass: assign remaining dogs randomly
for dog_info in dogs_data:
breed = random.choice(breeds)
create_dog(dog_info, breed.id)
# Commit all the changes
db.session.commit()
# Verify the seeding
dog_count = Dog.query.count()
print(f"Successfully seeded {dog_count} dogs to the database.")
# Print distribution of dogs across breeds
for breed in breeds:
count = breed_counts[breed.id]
print(f"Breed '{breed.name}': {count} dogs")
def seed_database():
"""Run all seeding functions in the correct order"""
create_breeds()
create_dogs()
if __name__ == '__main__':
seed_database()