mirror of
				https://github.com/langgenius/dify.git
				synced 2025-10-31 10:53:02 +00:00 
			
		
		
		
	
		
			
				
	
	
		
			72 lines
		
	
	
		
			1.7 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
			
		
		
	
	
			72 lines
		
	
	
		
			1.7 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
| from enum import Enum
 | |
| 
 | |
| from sqlalchemy import CHAR, TypeDecorator
 | |
| from sqlalchemy.dialects.postgresql import UUID
 | |
| 
 | |
| 
 | |
| class CreatedByRole(Enum):
 | |
|     """
 | |
|     Enum class for createdByRole
 | |
|     """
 | |
|     ACCOUNT = "account"
 | |
|     END_USER = "end_user"
 | |
| 
 | |
|     @classmethod
 | |
|     def value_of(cls, value: str) -> 'CreatedByRole':
 | |
|         """
 | |
|         Get value of given mode.
 | |
| 
 | |
|         :param value: mode value
 | |
|         :return: mode
 | |
|         """
 | |
|         for role in cls:
 | |
|             if role.value == value:
 | |
|                 return role
 | |
|         raise ValueError(f'invalid createdByRole value {value}')
 | |
| 
 | |
| 
 | |
| class CreatedFrom(Enum):
 | |
|     """
 | |
|     Enum class for createdFrom
 | |
|     """
 | |
|     SERVICE_API = "service-api"
 | |
|     WEB_APP = "web-app"
 | |
|     EXPLORE = "explore"
 | |
| 
 | |
|     @classmethod
 | |
|     def value_of(cls, value: str) -> 'CreatedFrom':
 | |
|         """
 | |
|         Get value of given mode.
 | |
| 
 | |
|         :param value: mode value
 | |
|         :return: mode
 | |
|         """
 | |
|         for role in cls:
 | |
|             if role.value == value:
 | |
|                 return role
 | |
|         raise ValueError(f'invalid createdFrom value {value}')
 | |
| 
 | |
| 
 | |
| class StringUUID(TypeDecorator):
 | |
|     impl = CHAR
 | |
|     cache_ok = True
 | |
| 
 | |
|     def process_bind_param(self, value, dialect):
 | |
|         if value is None:
 | |
|             return value
 | |
|         elif dialect.name == 'postgresql':
 | |
|             return str(value)
 | |
|         else:
 | |
|             return value.hex
 | |
| 
 | |
|     def load_dialect_impl(self, dialect):
 | |
|         if dialect.name == 'postgresql':
 | |
|             return dialect.type_descriptor(UUID())
 | |
|         else:
 | |
|             return dialect.type_descriptor(CHAR(36))
 | |
| 
 | |
|     def process_result_value(self, value, dialect):
 | |
|         if value is None:
 | |
|             return value
 | |
|         return str(value)
 | 
