FilesExpand file tree

 master

/

main.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

175 lines (151 loc) · 6.1 KB

 master

/

main.py

Copy path

Top

File metadata and controls

• Code
• Blame

175 lines (151 loc) · 6.1 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

# Copyright 2018 Google LLC

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

# http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

import datetime

import logging

import os

from flask import Flask, render_template, request, Response

import sqlalchemy

# Remember - storing secrets in plaintext is potentially unsafe. Consider using

# something like https://cloud.google.com/kms/ to help keep secrets secret.

db_user = os.environ.get("DB_USER")

db_pass = os.environ.get("DB_PASS")

db_name = os.environ.get("DB_NAME")

cloud_sql_connection_name = os.environ.get("CLOUD_SQL_CONNECTION_NAME")

app = Flask(__name__)

logger = logging.getLogger()

# [START cloud_sql_postgres_sqlalchemy_create]

# The SQLAlchemy engine will help manage interactions, including automatically

# managing a pool of connections to your database

db = sqlalchemy.create_engine(

# Equivalent URL:

# postgres+pg8000://<db_user>:<db_pass>@/<db_name>?unix_sock=/cloudsql/<cloud_sql_instance_name>/.s.PGSQL.5432

sqlalchemy.engine.url.URL(

drivername='postgres+pg8000',

username=db_user,

password=db_pass,

database=db_name,

query={

'unix_sock': '/cloudsql/{}/.s.PGSQL.5432'.format(

cloud_sql_connection_name)

}

),

# ... Specify additional properties here.

# [START_EXCLUDE]

# [START cloud_sql_postgres_sqlalchemy_limit]

# Pool size is the maximum number of permanent connections to keep.

pool_size=5,

# Temporarily exceeds the set pool_size if no connections are available.

max_overflow=2,

# The total number of concurrent connections for your application will be

# a total of pool_size and max_overflow.

# [END cloud_sql_postgres_sqlalchemy_limit]

# [START cloud_sql_postgres_sqlalchemy_backoff]

# SQLAlchemy automatically uses delays between failed connection attempts,

# but provides no arguments for configuration.

# [END cloud_sql_postgres_sqlalchemy_backoff]

# [START cloud_sql_postgres_sqlalchemy_timeout]

# 'pool_timeout' is the maximum number of seconds to wait when retrieving a

# new connection from the pool. After the specified amount of time, an

# exception will be thrown.

pool_timeout=30, # 30 seconds

# [END cloud_sql_postgres_sqlalchemy_timeout]

# [START cloud_sql_postgres_sqlalchemy_lifetime]

# 'pool_recycle' is the maximum number of seconds a connection can persist.

# Connections that live longer than the specified amount of time will be

# reestablished

pool_recycle=1800, # 30 minutes

# [END cloud_sql_postgres_sqlalchemy_lifetime]

# [END_EXCLUDE]

)

# [END cloud_sql_postgres_sqlalchemy_create]

@app.before_first_request

def create_tables():

# Create tables (if they don't already exist)

with db.connect() as conn:

conn.execute(

"CREATE TABLE IF NOT EXISTS votes "

"( vote_id SERIAL NOT NULL, time_cast timestamp NOT NULL, "

"candidate VARCHAR(6) NOT NULL, PRIMARY KEY (vote_id) );"

)

@app.route('/', methods=['GET'])

def index():

votes = []

with db.connect() as conn:

# Execute the query and fetch all results

recent_votes = conn.execute(

"SELECT candidate, time_cast FROM votes "

"ORDER BY time_cast DESC LIMIT 5"

).fetchall()

# Convert the results into a list of dicts representing votes

for row in recent_votes:

votes.append({

'candidate': row[0],

'time_cast': row[1]

})

stmt = sqlalchemy.text(

"SELECT COUNT(vote_id) FROM votes WHERE candidate=:candidate")

# Count number of votes for tabs

tab_result = conn.execute(stmt, candidate="TABS").fetchone()

tab_count = tab_result[0]

# Count number of votes for spaces

space_result = conn.execute(stmt, candidate="SPACES").fetchone()

space_count = space_result[0]

return render_template(

'index.html',

recent_votes=votes,

tab_count=tab_count,

space_count=space_count

)

@app.route('/', methods=['POST'])

def save_vote():

# Get the team and time the vote was cast.

team = request.form['team']

time_cast = datetime.datetime.utcnow()

# Verify that the team is one of the allowed options

if team != "TABS" and team != "SPACES":

logger.warning(team)

return Response(

response="Invalid team specified.",

status=400

)

# [START cloud_sql_postgres_sqlalchemy_connection]

# Preparing a statement before hand can help protect against injections.

stmt = sqlalchemy.text(

"INSERT INTO votes (time_cast, candidate)"

" VALUES (:time_cast, :candidate)"

)

try:

# Using a with statement ensures that the connection is always released

# back into the pool at the end of statement (even if an error occurs)

with db.connect() as conn:

conn.execute(stmt, time_cast=time_cast, candidate=team)

except Exception as e:

# If something goes wrong, handle the error in this section. This might

# involve retrying or adjusting parameters depending on the situation.

# [START_EXCLUDE]

logger.exception(e)

return Response(

status=500,

response="Unable to successfully cast vote! Please check the "

"application logs for more details."

)

# [END_EXCLUDE]

# [END cloud_sql_postgres_sqlalchemy_connection]

return Response(

status=200,

response="Vote successfully cast for '{}' at time {}!".format(

team, time_cast)

)

if __name__ == '__main__':

app.run(host='127.0.0.1', port=8080, debug=True)