Coverage for .tox/coverage/lib/python3.11/site-packages/wuttamess/mysql.py: 100%

1# -*- coding: utf-8; -*- 

2################################################################################ 

3# 

4# WuttaMess -- Fabric Automation Helpers 

5# Copyright © 2024-2026 Lance Edgar 

6# 

7# This file is part of Wutta Framework. 

8# 

9# Wutta Framework is free software: you can redistribute it and/or modify it 

10# under the terms of the GNU General Public License as published by the Free 

11# Software Foundation, either version 3 of the License, or (at your option) any 

12# later version. 

13# 

14# Wutta Framework is distributed in the hope that it will be useful, but 

15# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or 

16# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for 

17# more details. 

18# 

19# You should have received a copy of the GNU General Public License along with 

20# Wutta Framework. If not, see <http://www.gnu.org/licenses/>. 

21# 

22################################################################################ 

23""" 

24MySQL DB utilities 

25""" 

26 

27 

28def sql(c, sql_, database="", **kwargs): 

29 """ 

30 Execute some SQL. 

31 

32 :param c: Fabric connection. 

33 

34 :param sql_: SQL string to execute. 

35 

36 :param database: Name of the database on which to execute the SQL. 

37 

38 :returns: Result of the ``c.run()`` call (``mysql -e`` command). 

39 """ 

40 # some crazy quoting required here, see also 

41 # http://stackoverflow.com/a/1250279 

42 sql_ = sql_.replace("'", "'\"'\"'") 

43 

44 return c.run( 

45 f"mysql --execute='{sql_}' --batch --skip-column-names {database}", **kwargs 

46 ) 

47 

48 

49def user_exists(c, name, host="localhost"): 

50 """ 

51 Determine if a given MySQL user exists. 

52 

53 :param c: Fabric connection. 

54 

55 :param name: Username to check for. 

56 

57 :param host: Host portion for the MySQL user account; default is 

58 ``localhost``. 

59 

60 :returns: ``True`` if user exists, else ``False``. 

61 """ 

62 user = sql( 

63 c, 

64 f"SELECT User FROM user WHERE User = '{name}' and Host = '{host}'", 

65 database="mysql", 

66 ).stdout.strip() 

67 return user == name 

68 

69 

70def create_user(c, name, host="localhost", password=None, checkfirst=True): 

71 """ 

72 Create a MySQL user account. 

73 

74 :param c: Fabric connection. 

75 

76 :param name: Username to create. 

77 

78 :param host: Host portion for the MySQL user account; default is 

79 ``localhost``. 

80 

81 :param password: Optional password for the new user. If set, will 

82 call :func:`set_user_password()`. 

83 

84 :param checkfirst: If true (the default), first call 

85 :func:`user_exists()` and skip creating if already present. If 

86 false, then try to create user with no check. 

87 """ 

88 if not checkfirst or not user_exists(c, name, host): 

89 sql(c, f"CREATE USER '{name}'@'{host}';") 

90 if password: 

91 set_user_password(c, name, password, host=host) 

92 

93 

94def set_user_password(c, name, password, host="localhost"): 

95 """ 

96 Set the password for a MySQL user account. 

97 

98 :param c: Fabric connection. 

99 

100 :param name: Username whose password is to be set. 

101 

102 :param password: New password for the user. 

103 

104 :param host: Host portion for the MySQL user account; default is 

105 ``localhost``. 

106 """ 

107 # supposedly this is the new way to do it.. 

108 result = sql( 

109 c, 

110 f"ALTER USER '{name}'@'{host}' IDENTIFIED BY '{password}';", 

111 echo=False, 

112 hide=True, 

113 warn=True, 

114 ) 

115 if result.failed: # ..but it may fail for older systems, 

116 # in which case we try it the old way 

117 sql( 

118 c, 

119 f"SET PASSWORD FOR '{name}'@'{host}' = PASSWORD('{password}');", 

120 echo=False, 

121 hide=True, 

122 ) 

123 

124 

125def db_exists(c, name): 

126 """ 

127 Determine if a given MySQL database exists. 

128 

129 :param c: Fabric connection. 

130 

131 :param name: Name of the database to check for. 

132 

133 :returns: ``True`` if database exists, else ``False``. 

134 """ 

135 db = sql( 

136 c, 

137 f"SELECT SCHEMA_NAME FROM SCHEMATA WHERE SCHEMA_NAME = '{name}'", 

138 database="information_schema", 

139 ).stdout.strip() 

140 return db == name 

141 

142 

143def create_db(c, name, checkfirst=True, user=None): 

144 """ 

145 Create a MySQL database. 

146 

147 :param c: Fabric connection. 

148 

149 :param name: Name of the database to create. 

150 

151 :param checkfirst: If true (the default), first call 

152 :func:`db_exists()` and skip creating if already present. If 

153 false, then try to create DB with no check. 

154 

155 :param user: Optional user which should be granted full access to 

156 the database. If specified, should include username *and* 

157 host, e.g. ``poser@localhost``. See also 

158 :meth:`grant_access()`. 

159 """ 

160 if not checkfirst or not db_exists(c, name): 

161 c.run(f"mysqladmin create {name}") 

162 if user: 

163 grant_access(c, name, user) 

164 

165 

166def grant_access(c, dbname, user): 

167 """ 

168 Grant full access to the given database for the given user. 

169 

170 :param c: Fabric connection. 

171 

172 :param dbname: Name of the database to which access should be granted. 

173 

174 :param user: MySQL user which should be granted access. Must 

175 specify username *and* host, e.g. ``poser@localhost``. 

176 """ 

177 sql(c, f"grant all on `{dbname}`.* to {user}") 

178 

179 

180def drop_db(c, name, checkfirst=True): 

181 """ 

182 Drop a MySQL database. 

183 

184 :param c: Fabric connection. 

185 

186 :param name: Name of the database to drop. 

187 

188 :param checkfirst: If true (the default), first call 

189 :func:`db_exists()` and skip dropping if not present. If 

190 false, then try to drop DB with no check. 

191 """ 

192 if not checkfirst or db_exists(c, name): 

193 c.run(f"mysqladmin drop --force {name}")