Implement full Database object for Glue get_database() - fix for #3571. (#3572)

* implement potential fix for #3571.

* freeze_time decorator not used in TEST_SERVER_MODE
This commit is contained in:
aaronfowles-emis 2021-01-11 13:10:18 +00:00 committed by GitHub
commit a2246df1a4
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 75 additions and 17 deletions

View file

@ -1,6 +1,7 @@
from __future__ import unicode_literals
import time
from datetime import datetime
from moto.core import BaseBackend, BaseModel
from moto.compat import OrderedDict
@ -20,11 +21,11 @@ class GlueBackend(BaseBackend):
def __init__(self):
self.databases = OrderedDict()
def create_database(self, database_name):
def create_database(self, database_name, database_input):
if database_name in self.databases:
raise DatabaseAlreadyExistsException()
database = FakeDatabase(database_name)
database = FakeDatabase(database_name, database_input)
self.databases[database_name] = database
return database
@ -68,10 +69,26 @@ class GlueBackend(BaseBackend):
class FakeDatabase(BaseModel):
def __init__(self, database_name):
def __init__(self, database_name, database_input):
self.name = database_name
self.input = database_input
self.created_time = datetime.utcnow()
self.tables = OrderedDict()
def as_dict(self):
return {
"Name": self.name,
"Description": self.input.get("Description"),
"LocationUri": self.input.get("LocationUri"),
"Parameters": self.input.get("Parameters"),
"CreateTime": self.created_time.isoformat(),
"CreateTableDefaultPermissions": self.input.get(
"CreateTableDefaultPermissions"
),
"TargetDatabase": self.input.get("TargetDatabase"),
"CatalogId": self.input.get("CatalogId"),
}
class FakeTable(BaseModel):
def __init__(self, database_name, table_name, table_input):

View file

@ -21,19 +21,20 @@ class GlueResponse(BaseResponse):
return json.loads(self.body)
def create_database(self):
database_name = self.parameters["DatabaseInput"]["Name"]
self.glue_backend.create_database(database_name)
database_input = self.parameters.get("DatabaseInput")
database_name = database_input.get("Name")
self.glue_backend.create_database(database_name, database_input)
return ""
def get_database(self):
database_name = self.parameters.get("Name")
database = self.glue_backend.get_database(database_name)
return json.dumps({"Database": {"Name": database.name}})
return json.dumps({"Database": database.as_dict()})
def get_databases(self):
database_list = self.glue_backend.get_databases()
return json.dumps(
{"DatabaseList": [{"Name": database.name} for database in database_list]}
{"DatabaseList": [database.as_dict() for database in database_list]}
)
def create_table(self):